From 7470f9815585bef83db3add05a10bd8a545fa175 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Fri, 8 May 2026 13:25:52 +0000 Subject: [PATCH 1/9] fix(concurrency): add thread safety to InvariantService Add threading.RLock to InvariantService to protect shared state (_invariants dict, _enforcement_records list) from concurrent access by multiple threads during parallel plan execution. Prevents RuntimeError: dictionary changed size during iteration and data corruption in multi-threaded environments. Changes: - Added self._lock = RLock() in __init__ - Wrapped all public methods (add_invariant, list_invariants, remove_invariant, get_effective_invariants, enforce_invariants) with lock acquisition via context managers - Added helper read methods: get_enforcement_records(), get_invariant(), get_invariants_snapshot() -- all thread-safe - Added BDD tests for concurrent access patterns - Updated CHANGELOG.md and CONTRIBUTORS.md ISSUES CLOSED: #7524 --- CHANGELOG.md | 10 + CONTRIBUTORS.md | 1 + .../invariant_service_thread_safety.feature | 75 +++ .../invariant_service_thread_safety_steps.py | 435 ++++++++++++++++++ .../application/services/invariant_service.py | 81 +++- 5 files changed, 589 insertions(+), 13 deletions(-) create mode 100644 features/invariant_service_thread_safety.feature create mode 100644 features/steps/invariant_service_thread_safety_steps.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d63d89c32..72bae9e50 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -340,6 +340,16 @@ ensuring data is stored with proper parameter values. a `_TextualPromptInput` composite (Horizontal + Static + Input) when Textual is available, and a `_FallbackPromptInput` otherwise. Zero `# type: ignore` suppressions — all typing uses Protocol definitions and `cast()`. +- **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). + - **Actor CLI NAME argument made optional, derived from YAML config** (#4186): The `agents actor add` positional `NAME` argument is now optional (defaults to `None`). When omitted, the actor name is derived from the `name` field in diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index 37bd8c5ec..f266b1fb3 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..abe44ca9d --- /dev/null +++ b/features/invariant_service_thread_safety.feature @@ -0,0 +1,75 @@ +# This test captures the concurrency bug discovered in issue #7524. +# +# InvariantService stored invariants (_invariants dict) and enforcement records +# (_enforcement_records list) without any threading.Lock. Concurrent calls from +# parallel plan execution could raise RuntimeError: dictionary changed size during +# iteration or silently corrupt data. +# +# The fix adds a threading.RLock to InvariantService guarding all shared mutable +# state (see commit that implements the fix). + +@tdd_issue @tdd_issue_7524 +Feature: TDD Issue #7524 - InvariantService Thread Safety + InvariantService must be thread-safe so that concurrent calls from parallel + plan execution threads cannot race on _invariants or _enforcement_records. + + Scenario Outline: Concurrent add_invariant() must not corrupt the dict + Given an invariant service + When threads concurrently add d invariants through a barrier + And the service has no RuntimeError raised during concurrent access + Then the service should have exactly d active invariants + And all added invariants should be retrievable via get_invariant_snapshot + + Examples: + | n | count | total | + | 2 | 5 | 10 | + | 4 | 5 | 20 | + | 8 | 3 | 24 | + | 10| 2 | 20 | + + Scenario Outline: Concurrent list_invariants() must not raise during iteration + Given an invariant service with d pre-existing invariants + When 5 threads concurrently list all invariants through a barrier + And no thread raised RuntimeError during concurrent reads + Then the caller should receive a valid list of at least d active invariants + + Examples: + | count | min | + | 10 | 10 | + | 20 | 20 | + | 5 | 5 | + + Scenario Outline: Concurrent remove_invariant() must not race with add + Given an invariant service + When threads concurrently add invariants while concurrent threads remove different ones through a barrier + And no thread raised RuntimeError during mixed access + Then the service should have a consistent count of active invariants + + Examples: + | n | m | + | 5 | 2 | + | 4 | 3 | + | 8 | 4 | + + Scenario Outline: Concurrent enforce_invariants() must not corrupt the record list + Given an invariant service with d invariants to enforce + When threads concurrently call enforce_invariants on the same set through a barrier + And no thread raised RuntimeError during enforcement + Then each thread should have received exactly d enforcement records + + Examples: + | count | n | + | 3 | 4 | + | 5 | 8 | + | 2 | 10| + + Scenario Outline: Concurrent enforce and list must both succeed + Given an invariant service with d invariants already enforced by a previous thread + When threads concurrently enforce new invariants while other threads list them through a barrier + And no thread raised RuntimeError during mixed enforcement and listing + Then the enforcement record count should be consistent across all threads + + Examples: + | count | n | + | 3 | 4 | + | 5 | 6 | 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..7d8bc70d1 --- /dev/null +++ b/features/steps/invariant_service_thread_safety_steps.py @@ -0,0 +1,435 @@ +"""Step definitions for invariant_service_thread_safety.feature. + +Tests thread safety of InvariantService by launching multiple threads that +perform concurrent operations on a shared InvariantService instance, verifying +that no RuntimeError (dictionary changed size during iteration) is raised and +that data remains consistent across all concurrent accesses. + +See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/7524 +""" + +from __future__ import annotations + +import threading +from typing import cast + +from behave import given, then, when # type: ignore[import-untyped] +from behave.runner import Context + +from cleveragents.application.services.invariant_service import InvariantService +from cleveragents.domain.models.core.invariant import InvariantScope + + +# ================================================================ +# Setup helpers +# ================================================================ + + +def _create_barrier_and_threads(n: int) -> tuple[threading.Barrier, list[threading.Thread]]: + """Create a barrier for n threads and an empty thread list.""" + barrier = threading.Barrier(n) + threads: list[threading.Thread] = [] + return barrier, threads + + +# ================================================================ +# Thread-safety state tracking per-thread +# ================================================================ + + +@given("an invariant service") +def step_service(context: Context) -> None: + """Create a fresh InvariantService with empty concurrency tracking.""" + context.service = InvariantService() + context.concurrent_errors: list[Exception] = [] + context.concurrent_results: dict[str, object] = {} + + +@given("an invariant service with {count:d} pre-existing invariants") +def step_service_initialized(context: Context, count: int) -> None: + """Create an InvariantService already populated with {count} invariants.""" + context.service = InvariantService() + for i in range(count): + context.service.add_invariant( + text=f"Constraint {i + 1}", + scope=InvariantScope.GLOBAL, + source_name="test-context", + ) + context.concurrent_errors = [] + context.concurrent_results = {} + + +@given("an invariant service with {count:d} invariants to enforce") +def step_service_with_invariants(context: Context, count: int) -> None: + """Create an InvariantService populated with specific invariants for enforcement.""" + context.service = InvariantService() + for i in range(count): + inv = context.service.add_invariant( + text=f"Enforced constraint {i + 1}", + scope=InvariantScope.GLOBAL, + source_name="enforcement-test", + ) + # Store the invariant IDs so enforcement scenarios can reference them + context.enforcement_inv_ids = list(context.concurrent_results) if hasattr(context, "concurrent_results") else [] + context.concurrent_errors = [] + context.concurrent_results = {} + + +@given("an invariant service with {count:d} invariants already enforced by a previous thread") +def step_service_with_enforced_invariants(context: Context, count: int) -> None: + """Create an InvariantService where enforcement records already exist.""" + context.service = InvariantService() + invariants = [] + for i in range(count): + inv = context.service.add_invariant( + text=f"Already enforced {i + 1}", + scope=InvariantScope.GLOBAL, + source_name="pre-enforced", + ) + invariants.append(inv) + context.service.enforce_invariants( + plan_id="previous-plan", + invariants=invariants, + ) + context.concurrent_errors = [] + context.concurrent_results = {} + + +# ================================================================ +# Concurrent add_invariant() — Scenario: concurrent adds +# ================================================================ + + +@when( + " threads concurrently add d invariants through a barrier", +) +def step_concurrent_add(context: Context, n: int, count: int) -> None: + """Launch n thread groups, each adding {count} invariants at once.""" + total = n * count # actual total added by this scenario + barrier, threads = _create_barrier_and_threads(n) + + def worker(thread_index: int) -> None: + try: + barrier.wait(timeout=10) + for i in range(count): + context.service.add_invariant( + text=f"Concurrent add T{thread_index}-{i}", + scope=InvariantScope.GLOBAL, + source_name="concurrent-test", + ) + except Exception as exc: + with context._error_mutex if hasattr(context, "_error_mutex") else threading.Lock(): + context.concurrent_errors.append(exc) + + for idx in range(n): + t = threading.Thread(target=worker, args=(idx,), daemon=True) + threads.append(t) + t.start() + + # Also create a mutex for the first call since it may not exist yet + if not hasattr(context, "_error_mutex"): + context._error_mutex = threading.Lock() + + for t in threads: + t.join(timeout=30) + + +@then("the service has no RuntimeError raised during concurrent access") +def step_no_runtime_error_adds(context: Context) -> None: + """Assert none of the worker threads hit a RuntimeError.""" + runtime_errors = [ + e for e in context.concurrent_errors if isinstance(e, RuntimeError) + ] + assert not runtime_errors, ( + f"RuntimeError raised during concurrent adds: {runtime_errors}. " + "The RLock in InvariantService is likely missing or ineffective." + ) + + +@then("the service should have exactly {total:d} active invariants") +def step_concurrent_add_count(context: Context, total: int) -> None: + """Verify the expected number of invariants was added (10 from 2x5 etc.).""" + actual = context.service.list_invariants() + assert len(actual) == total, ( + f"Expected {total} active invariants after concurrent adds, " + f"but got {len(actual)}. Data loss or corruption detected." + ) + + +@then("all added invariants should be retrievable via get_invariant_snapshot") +def step_snapshot_contains_all(context: Context) -> None: + """Confirm get_invariants_snapshot returns consistent data.""" + snapshot = context.service.get_invariants_snapshot() + listed = context.service.list_invariants() + assert len(snapshot) == len(listed), ( + f"Snapshot has {len(snapshot)} invariants but list_invariants found " + f"{len(listed)}. Inconsistent reads detected." + ) + + +# ================================================================ +# Concurrent list_invariants() — Scenario: concurrent reads +# ================================================================ + + +@when("5 threads concurrently list all invariants through a barrier") +def step_concurrent_list(context: Context) -> None: + """Launch 5 threads that each call list_invariants simultaneously.""" + barrier = threading.Barrier(5) + context.concurrent_errors = [] + context._list_results: list[list] = [] + + def worker() -> None: + try: + barrier.wait(timeout=10) + results = context.service.list_invariants() + with getattr(context, "_list_lock", threading.Lock()): + context._list_results.append(results) + except Exception as exc: + context.concurrent_errors.append(exc) + + # Initialize lock if not set + if not hasattr(context, "_list_lock"): + context._list_lock = threading.Lock() + + threads = [threading.Thread(target=worker, daemon=True) for _ in range(5)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=30) + + +@then("no thread raised RuntimeError during concurrent reads") +def step_no_runtime_error_lists(context: Context) -> None: + """Verify no thread hit a RuntimeError during list_invariants().""" + runtime_errors = [e for e in context.concurrent_errors if isinstance(e, RuntimeError)] + assert not runtime_errors, ( + f"RuntimeError raised during concurrent reads: {runtime_errors}. " + "The RLock is missing in list_invariants()." + ) + + +@then("the caller should receive a valid list of at least {min:d} active invariants") +def step_list_produces_valid_result(context: Context, min: int) -> None: + """Each thread's list result should be consistent.""" + # All threads should have seen the same count (snapshot consistency) + for results in context._list_results: + assert len(results) >= min, ( + f"Thread saw only {len(results)} active invariants " + f"(expected at least {min}). " + "A reader may have seen a partially-written dict." + ) + + +# ================================================================ +# Concurrent remove + add — Scenario: mixed access +# ================================================================ + + +@when( + " threads concurrently add invariants while concurrent " + "threads remove different ones through a barrier", +) +def step_concurrent_mixed(context: Context, n: int, m: int) -> None: + """Launch {n} add-threads and {m} remove-threads simultaneously.""" + total_workers = n + m + barrier, threads = _create_barrier_and_threads(total_workers) + + def add_worker(idx: int) -> None: + try: + barrier.wait(timeout=10) + context.service.add_invariant( + text=f"Concurrent add mixed-{idx}", + scope=InvariantScope.GLOBAL, + source_name="mixed-test", + ) + except Exception as exc: + with getattr(context, "_mix_lock", threading.Lock()): + context.concurrent_errors.append(exc) + + def remove_worker(idx: int) -> None: + try: + barrier.wait(timeout=10) + # Get some invariants to remove. Use snapshot to ensure list is stable. + invs = context.service.list_invariants() + if len(invs) > 0: + target_id = invs[idx % len(invs)].id + try: + context.service.remove_invariant(target_id) + except Exception: + pass # already removed by another worker - expected race + except Exception as exc: + with getattr(context, "_mix_lock", threading.Lock()): + context.concurrent_errors.append(exc) + + if not hasattr(context, "_mix_lock"): + context._mix_lock = threading.Lock() + context.concurrent_errors = [] + + for i in range(n): + threads.append(threading.Thread(target=add_worker, args=(i,), daemon=True)) + remove_workers = n # index offset into the list; add workers occupy first `n` slots + for i in range(m): + threads.append(threading.Thread(target=remove_worker, args=(i,), daemon=True)) + + for t in threads: + t.start() + for t in threads: + t.join(timeout=30) + + +@then("no thread raised RuntimeError during mixed access") +def step_no_runtime_error_mixed(context: Context) -> None: + """Verify no RuntimeError was thrown during concurrent add + remove.""" + runtime_errors = [e for e in context.concurrent_errors if isinstance(e, RuntimeError)] + assert not runtime_errors, ( + f"RuntimeError raised during mixed access: {runtime_errors}. " + "The RLock is missing or ineffective for concurrent read+write." + ) + + +@then("the service should have a consistent count of active invariants") +def step_consistent_mixed_count(context: Context) -> None: + """Verify we can consistently query the service after mixed operations.""" + try: + _ = context.service.list_invariants() + _ = context.service.get_invariants_snapshot() + except RuntimeError as exc: + raise AssertionError( + f"Inconsistent state detected after mixed concurrent access: {exc}" + ) from exc + + +# ================================================================ +# Concurrent enforce_invariants() — Scenario: concurrent enforcement +# ================================================================ + + +@when( + " threads concurrently call enforce_invariants on the same set " + "through a barrier", +) +def step_concurrent_enforce(context: Context, n: int) -> None: + """Launch {n} threads that all enforce the same invariants simultaneously.""" + barrier = threading.Barrier(n) + context.concurrent_errors = [] + context._enforcement_results: dict[int, list] = {} + + # Re-read invariants fresh inside each thread to get stable snapshot. + invariants = context.service.list_invariants() + + def worker(thread_idx: int) -> None: + try: + barrier.wait(timeout=10) + # Each thread re-reads from snapshot under lock (via the service) + invs = context.service.list_invariants() + records = context.service.enforce_invariants( + plan_id=f"plan-thread-{thread_idx}", + invariants=invs, + ) + with getattr(context, "_enf_lock", threading.Lock()): + context._enforcement_results[thread_idx] = records + except Exception as exc: + context.concurrent_errors.append(exc) + + if not hasattr(context, "_enf_lock"): + context._enf_lock = threading.Lock() + + threads = [threading.Thread(target=worker, args=(i,), daemon=True) for i in range(n)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=30) + + +@then("each thread should have received exactly {count:d} enforcement records") +def step_enforce_record_count(context: Context, count: int) -> None: + """Every thread's enforcement result list length must match expected count.""" + for idx, records in context._enforcement_results.items(): + assert len(records) == count, ( + f"Thread {idx} received {len(records)} enforcement records, " + f"expected {count}. Concurrent enforcement corrupted the result." + ) + + +@then("no thread raised RuntimeError during enforcement") +def step_no_runtime_error_enforce(context: Context) -> None: + """Verify no RuntimeError during concurrent enforce_invariants().""" + runtime_errors = [e for e in context.concurrent_errors if isinstance(e, RuntimeError)] + assert not runtime_errors, ( + f"RuntimeError raised during enforcement: {runtime_errors}. " + "_enforcement_records.extend() is likely unprotected." + ) + + +# ================================================================ +# Mixed enforce + list — Scenario: concurrent enforcement and listing +# ================================================================ + + +@when( + " threads concurrently enforce new invariants while other threads " + "list them through a barrier", +) +def step_concurrent_enforce_list(context: Context, n: int) -> None: + """Launch {n} enforcers and {n} lister workers simultaneously.""" + total_workers = n + n # enforcers + listers + barrier, threads = _create_barrier_and_threads(total_workers) + context.concurrent_errors = [] + + invariants = context.service.list_invariants() + + def enforce_worker(idx: int) -> None: + try: + barrier.wait(timeout=10) + invs = context.service.list_invariants() + context.service.enforce_invariants( + plan_id=f"enforce-worker-{idx}", + invariants=invs, + ) + except Exception as exc: + with getattr(context, "_mix2_lock", threading.Lock()): + context.concurrent_errors.append(exc) + + def list_worker(idx: int) -> None: + try: + barrier.wait(timeout=10) + _ = context.service.list_invariants() + except Exception as exc: + with getattr(context, "_mix2_lock", threading.Lock()): + context.concurrent_errors.append(exc) + + if not hasattr(context, "_mix2_lock"): + context._mix2_lock = threading.Lock() + + for i in range(n): + threads.append(threading.Thread(target=enforce_worker, args=(i,), daemon=True)) + for i in range(n): + threads.append(threading.Thread(target=list_worker, args=(i,), daemon=True)) + + for t in threads: + t.start() + for t in threads: + t.join(timeout=30) + + +@then("no thread raised RuntimeError during mixed enforcement and listing") +def step_no_runtime_error_mixed_enforce(context: Context) -> None: + """Verify no RuntimeError during concurrent enforce + list.""" + runtime_errors = [e for e in context.concurrent_errors if isinstance(e, RuntimeError)] + assert not runtime_errors, ( + f"RuntimeError raised during mixed enforcement and listing: {runtime_errors}" + ) + + +@then("the enforcement record count should be consistent across all threads") +def step_enforcement_records_consistent(context: Context) -> None: + """Verify the final state is readable without errors.""" + try: + _ = context.service.get_enforcement_records() + _ = context.service.list_invariants() + _ = context.service.get_invariants_snapshot() + except RuntimeError as exc: + raise AssertionError( + f"Inconsistent state after concurrent enforce + list: {exc}" + ) from exc diff --git a/src/cleveragents/application/services/invariant_service.py b/src/cleveragents/application/services/invariant_service.py index 10e587ce8..9126acf7a 100644 --- a/src/cleveragents/application/services/invariant_service.py +++ b/src/cleveragents/application/services/invariant_service.py @@ -14,11 +14,19 @@ 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 + +A ``threading.RLock`` (re-entrant lock) guards all shared mutable state +(``_invariants`` dict, ``_enforcement_records`` list). Every public method +acquires the lock before reading or writing shared data; the lock is re-entrant +so that nested calls from within a locked block do **not** deadlock. + 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 @@ -50,6 +58,10 @@ class InvariantService: Provides add, list, remove (soft-delete), effective-set computation, and enforcement record creation. All storage is in-memory. + + Thread safety is provided via a ``threading.RLock`` that protects all + shared state mutations so that concurrent readers and writers cannot + race or raise ``RuntimeError: dictionary changed size during iteration``. """ def __init__(self, event_bus: EventBus | None = None) -> None: @@ -60,6 +72,7 @@ class InvariantService: """ self._invariants: dict[str, Invariant] = {} self._enforcement_records: list[InvariantEnforcementRecord] = [] + self._lock = RLock() # guards _invariants and _enforcement_records self._logger = logger.bind(service="invariant") self._sanitizer = PromptSanitizer() self._event_bus = event_bus @@ -99,7 +112,9 @@ class InvariantService: source_name=source_name.strip(), ) - self._invariants[invariant.id] = invariant + with self._lock: + self._invariants[invariant.id] = invariant + self._logger.info( "Invariant added", invariant_id=invariant.id, @@ -132,7 +147,8 @@ class InvariantService: project_name=source_name if scope == InvariantScope.PROJECT else None, ) - result = [inv for inv in self._invariants.values() if inv.active] + with self._lock: + result = [inv for inv in self._invariants.values() if inv.active] if scope is not None: result = [inv for inv in result if inv.scope == scope] @@ -157,15 +173,17 @@ class InvariantService: if not invariant_id or not invariant_id.strip(): raise ValidationError("Invariant ID must not be empty") - inv = self._invariants.get(invariant_id) - if inv is None: - raise NotFoundError( - resource_type="invariant", - resource_id=invariant_id, - ) - # Invariant is frozen (immutable); create a new instance with active=False - deactivated = inv.model_copy(update={"active": False}) - self._invariants[invariant_id] = deactivated + with self._lock: + inv = self._invariants.get(invariant_id) + if inv is None: + raise NotFoundError( + resource_type="invariant", + resource_id=invariant_id, + ) + # Invariant is frozen (immutable); create a new instance with active=False + deactivated = inv.model_copy(update={"active": False}) + self._invariants[invariant_id] = deactivated + self._logger.info("Invariant removed (soft-delete)", invariant_id=invariant_id) return deactivated @@ -194,7 +212,8 @@ class InvariantService: Returns: Merged, de-duplicated list of effective invariants. """ - active = [inv for inv in self._invariants.values() if inv.active] + with self._lock: + active = [inv for inv in self._invariants.values() if inv.active] plan_invs = [ inv @@ -467,7 +486,9 @@ class InvariantService: exc_info=True, ) - self._enforcement_records.extend(records) + with self._lock: + self._enforcement_records.extend(records) + self._logger.info( "Invariants enforced", plan_id=plan_id, @@ -513,3 +534,37 @@ class InvariantService: exc_info=True, ) return records + + # ------------------------------------------------------------------ + # Thread-safe read helpers + # ------------------------------------------------------------------ + + def get_enforcement_records(self) -> list[InvariantEnforcementRecord]: + """Return enforcement records (thread-safe read). + + Returns: + List of ``InvariantEnforcementRecord`` objects collected so far. + """ + with self._lock: + return list(self._enforcement_records) + + def get_invariant(self, invariant_id: str) -> Invariant | None: + """Return a single invariant by ID (thread-safe read). + + Args: + invariant_id: The ULID of the desired invariant. + + Returns: + The ``Invariant`` if found, else ``None``. + """ + with self._lock: + return self._invariants.get(invariant_id) + + def get_invariants_snapshot(self) -> dict[str, Invariant]: + """Return a snapshot copy of all invariants (thread-safe read). + + Returns: + A shallow copy of the internal ``_invariants`` dict. + """ + with self._lock: + return dict(self._invariants) -- 2.52.0 From 50098eada75fe12d4bcf6796f8122a07a038ab16 Mon Sep 17 00:00:00 2001 From: HAL9000 Date: Sat, 9 May 2026 10:23:45 +0000 Subject: [PATCH 2/9] fix(concurrency): add thread safety to InvariantService (lint and Behave fixes) Address all review feedback from HAL9001: - Remove unused 'from typing import cast' import (F401) - Replace unused variables with _ prefix (F841) - Use contextlib.suppress(Exception) instead of try/except/pass (SIM105) - Fix ScenarioOutline decorators to use curly-brace {param:d} syntax - Convert And-Then steps to proper Then/assertion steps for Behave compatibility - Rename branch from pr_fix/8209 to bugfix/m3-invariant-service-thread-safety Fixes: resolves PR #11051 review comments --- .../invariant_service_thread_safety.feature | 42 +++++++++---------- .../invariant_service_thread_safety_steps.py | 38 ++++++++--------- 2 files changed, 39 insertions(+), 41 deletions(-) diff --git a/features/invariant_service_thread_safety.feature b/features/invariant_service_thread_safety.feature index abe44ca9d..b36363a01 100644 --- a/features/invariant_service_thread_safety.feature +++ b/features/invariant_service_thread_safety.feature @@ -15,9 +15,9 @@ Feature: TDD Issue #7524 - InvariantService Thread Safety Scenario Outline: Concurrent add_invariant() must not corrupt the dict Given an invariant service - When threads concurrently add d invariants through a barrier - And the service has no RuntimeError raised during concurrent access - Then the service should have exactly d active invariants + When {n} threads concurrently add {count:d} invariants through a barrier + Then the service has no RuntimeError raised during concurrent access + And the service should have exactly {total:d} active invariants And all added invariants should be retrievable via get_invariant_snapshot Examples: @@ -28,10 +28,10 @@ Feature: TDD Issue #7524 - InvariantService Thread Safety | 10| 2 | 20 | Scenario Outline: Concurrent list_invariants() must not raise during iteration - Given an invariant service with d pre-existing invariants + Given an invariant service with {count:d} pre-existing invariants When 5 threads concurrently list all invariants through a barrier - And no thread raised RuntimeError during concurrent reads - Then the caller should receive a valid list of at least d active invariants + Then no thread raised RuntimeError during concurrent reads + And the caller should receive a valid list of at least {min:d} active invariants Examples: | count | min | @@ -41,9 +41,9 @@ Feature: TDD Issue #7524 - InvariantService Thread Safety Scenario Outline: Concurrent remove_invariant() must not race with add Given an invariant service - When threads concurrently add invariants while concurrent threads remove different ones through a barrier - And no thread raised RuntimeError during mixed access - Then the service should have a consistent count of active invariants + When {n} threads concurrently add invariants while {m:d} concurrent threads remove different ones through a barrier + Then no thread raised RuntimeError during mixed access + And the service should have a consistent count of active invariants Examples: | n | m | @@ -52,22 +52,22 @@ Feature: TDD Issue #7524 - InvariantService Thread Safety | 8 | 4 | Scenario Outline: Concurrent enforce_invariants() must not corrupt the record list - Given an invariant service with d invariants to enforce - When threads concurrently call enforce_invariants on the same set through a barrier - And no thread raised RuntimeError during enforcement - Then each thread should have received exactly d enforcement records + Given an invariant service with {count:d} invariants to enforce + When {n:d} threads concurrently call enforce_invariants on the same set through a barrier + Then no thread raised RuntimeError during enforcement + And each thread should have received exactly {total:d} enforcement records Examples: - | count | n | - | 3 | 4 | - | 5 | 8 | - | 2 | 10| + | count | n | total | + | 3 | 4 | 3 | + | 5 | 8 | 5 | + | 2 | 10| 2 | Scenario Outline: Concurrent enforce and list must both succeed - Given an invariant service with d invariants already enforced by a previous thread - When threads concurrently enforce new invariants while other threads list them through a barrier - And no thread raised RuntimeError during mixed enforcement and listing - Then the enforcement record count should be consistent across all threads + Given an invariant service with {count:d} invariants already enforced by a previous thread + When {n:d} threads concurrently enforce new invariants while other threads list them through a barrier + Then no thread raised RuntimeError during mixed enforcement and listing + And the enforcement record count should be consistent across all threads Examples: | count | n | diff --git a/features/steps/invariant_service_thread_safety_steps.py b/features/steps/invariant_service_thread_safety_steps.py index 7d8bc70d1..56267263d 100644 --- a/features/steps/invariant_service_thread_safety_steps.py +++ b/features/steps/invariant_service_thread_safety_steps.py @@ -10,8 +10,8 @@ See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/7524 from __future__ import annotations +import contextlib import threading -from typing import cast from behave import given, then, when # type: ignore[import-untyped] from behave.runner import Context @@ -64,7 +64,7 @@ def step_service_with_invariants(context: Context, count: int) -> None: """Create an InvariantService populated with specific invariants for enforcement.""" context.service = InvariantService() for i in range(count): - inv = context.service.add_invariant( + _ = context.service.add_invariant( text=f"Enforced constraint {i + 1}", scope=InvariantScope.GLOBAL, source_name="enforcement-test", @@ -101,11 +101,11 @@ def step_service_with_enforced_invariants(context: Context, count: int) -> None: @when( - " threads concurrently add d invariants through a barrier", + "{n:d} threads concurrently add {count:d} invariants through a barrier", ) def step_concurrent_add(context: Context, n: int, count: int) -> None: - """Launch n thread groups, each adding {count} invariants at once.""" - total = n * count # actual total added by this scenario + """Launch n thread groups, each adding count invariants at once.""" + _ = n * count # actual total added by this scenario barrier, threads = _create_barrier_and_threads(n) def worker(thread_index: int) -> None: @@ -210,13 +210,13 @@ def step_no_runtime_error_lists(context: Context) -> None: @then("the caller should receive a valid list of at least {min:d} active invariants") -def step_list_produces_valid_result(context: Context, min: int) -> None: +def step_list_produces_valid_result(context: Context, min_: int) -> None: """Each thread's list result should be consistent.""" # All threads should have seen the same count (snapshot consistency) for results in context._list_results: - assert len(results) >= min, ( + assert len(results) >= min_, ( f"Thread saw only {len(results)} active invariants " - f"(expected at least {min}). " + f"(expected at least {min_}). " "A reader may have seen a partially-written dict." ) @@ -227,11 +227,11 @@ def step_list_produces_valid_result(context: Context, min: int) -> None: @when( - " threads concurrently add invariants while concurrent " + "{n:d} threads concurrently add invariants while {m:d} concurrent " "threads remove different ones through a barrier", ) def step_concurrent_mixed(context: Context, n: int, m: int) -> None: - """Launch {n} add-threads and {m} remove-threads simultaneously.""" + """Launch n add-threads and m remove-threads simultaneously.""" total_workers = n + m barrier, threads = _create_barrier_and_threads(total_workers) @@ -254,10 +254,8 @@ def step_concurrent_mixed(context: Context, n: int, m: int) -> None: invs = context.service.list_invariants() if len(invs) > 0: target_id = invs[idx % len(invs)].id - try: + with contextlib.suppress(Exception): context.service.remove_invariant(target_id) - except Exception: - pass # already removed by another worker - expected race except Exception as exc: with getattr(context, "_mix_lock", threading.Lock()): context.concurrent_errors.append(exc) @@ -268,7 +266,7 @@ def step_concurrent_mixed(context: Context, n: int, m: int) -> None: for i in range(n): threads.append(threading.Thread(target=add_worker, args=(i,), daemon=True)) - remove_workers = n # index offset into the list; add workers occupy first `n` slots + _ = n # index offset into the list; add workers occupy first `n` slots for i in range(m): threads.append(threading.Thread(target=remove_worker, args=(i,), daemon=True)) @@ -306,17 +304,17 @@ def step_consistent_mixed_count(context: Context) -> None: @when( - " threads concurrently call enforce_invariants on the same set " + "{n:d} threads concurrently call enforce_invariants on the same set " "through a barrier", ) def step_concurrent_enforce(context: Context, n: int) -> None: - """Launch {n} threads that all enforce the same invariants simultaneously.""" + """Launch n threads that all enforce the same invariants simultaneously.""" barrier = threading.Barrier(n) context.concurrent_errors = [] context._enforcement_results: dict[int, list] = {} # Re-read invariants fresh inside each thread to get stable snapshot. - invariants = context.service.list_invariants() + _ = context.service.list_invariants() def worker(thread_idx: int) -> None: try: @@ -368,16 +366,16 @@ def step_no_runtime_error_enforce(context: Context) -> None: @when( - " threads concurrently enforce new invariants while other threads " + "{n:d} threads concurrently enforce new invariants while other threads " "list them through a barrier", ) def step_concurrent_enforce_list(context: Context, n: int) -> None: - """Launch {n} enforcers and {n} lister workers simultaneously.""" + """Launch n enforcers and n lister workers simultaneously.""" total_workers = n + n # enforcers + listers barrier, threads = _create_barrier_and_threads(total_workers) context.concurrent_errors = [] - invariants = context.service.list_invariants() + _ = context.service.list_invariants() def enforce_worker(idx: int) -> None: try: -- 2.52.0 From b87827f351ff292afa4505f9c91f599bab11a3be Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Thu, 11 Jun 2026 04:00:06 -0400 Subject: [PATCH 3/9] chore: re-trigger CI [controller] -- 2.52.0 From 6ec7479c2ef088874358fcbf40198b0fa2a7d775 Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Wed, 17 Jun 2026 14:32:36 -0400 Subject: [PATCH 4/9] chore: re-trigger CI [controller] -- 2.52.0 From 957c6c622ebcd8244ee8c24325bda9c7551511e2 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Wed, 17 Jun 2026 16:23:11 -0400 Subject: [PATCH 5/9] fix(tests): correct ScenarioOutline syntax and step parameter mismatches Replace {col}/{col:d} curly-brace column refs with Behave-required angle-bracket syntax in all ScenarioOutline step text. Rename ambiguous given step to avoid duplicate step definition conflicts. Fix min_ -> min_count parameter mismatch, move _error_mutex init before thread start to eliminate lock race, apply ruff format wraps, and add get_invariant() call in snapshot step to cover invariant_service.py:560-561. ISSUES CLOSED: #7524 --- .../invariant_service_thread_safety.feature | 40 ++++++------- .../invariant_service_thread_safety_steps.py | 60 +++++++++++++------ 2 files changed, 61 insertions(+), 39 deletions(-) diff --git a/features/invariant_service_thread_safety.feature b/features/invariant_service_thread_safety.feature index b36363a01..b692a0f32 100644 --- a/features/invariant_service_thread_safety.feature +++ b/features/invariant_service_thread_safety.feature @@ -14,10 +14,10 @@ Feature: TDD Issue #7524 - InvariantService Thread Safety plan execution threads cannot race on _invariants or _enforcement_records. Scenario Outline: Concurrent add_invariant() must not corrupt the dict - Given an invariant service - When {n} threads concurrently add {count:d} invariants through a barrier + Given an invariant service for thread-safety testing + When threads concurrently add invariants through a barrier Then the service has no RuntimeError raised during concurrent access - And the service should have exactly {total:d} active invariants + And the service should have exactly active invariants And all added invariants should be retrievable via get_invariant_snapshot Examples: @@ -28,20 +28,20 @@ Feature: TDD Issue #7524 - InvariantService Thread Safety | 10| 2 | 20 | Scenario Outline: Concurrent list_invariants() must not raise during iteration - Given an invariant service with {count:d} pre-existing invariants + Given an invariant service with pre-existing invariants When 5 threads concurrently list all invariants through a barrier Then no thread raised RuntimeError during concurrent reads - And the caller should receive a valid list of at least {min:d} active invariants + And the caller should receive a valid list of at least active invariants Examples: - | count | min | - | 10 | 10 | - | 20 | 20 | - | 5 | 5 | + | count | min_count | + | 10 | 10 | + | 20 | 20 | + | 5 | 5 | Scenario Outline: Concurrent remove_invariant() must not race with add - Given an invariant service - When {n} threads concurrently add invariants while {m:d} concurrent threads remove different ones through a barrier + Given an invariant service for thread-safety testing + When threads concurrently add invariants while concurrent threads remove different ones through a barrier Then no thread raised RuntimeError during mixed access And the service should have a consistent count of active invariants @@ -52,20 +52,20 @@ Feature: TDD Issue #7524 - InvariantService Thread Safety | 8 | 4 | Scenario Outline: Concurrent enforce_invariants() must not corrupt the record list - Given an invariant service with {count:d} invariants to enforce - When {n:d} threads concurrently call enforce_invariants on the same set through a barrier + Given an invariant service with invariants to enforce + When threads concurrently call enforce_invariants on the same set through a barrier Then no thread raised RuntimeError during enforcement - And each thread should have received exactly {total:d} enforcement records + And each thread should have received exactly enforcement records Examples: - | count | n | total | - | 3 | 4 | 3 | - | 5 | 8 | 5 | - | 2 | 10| 2 | + | count | n | total | + | 3 | 4 | 3 | + | 5 | 8 | 5 | + | 2 | 10 | 2 | Scenario Outline: Concurrent enforce and list must both succeed - Given an invariant service with {count:d} invariants already enforced by a previous thread - When {n:d} threads concurrently enforce new invariants while other threads list them through a barrier + Given an invariant service with invariants already enforced by a previous thread + When threads concurrently enforce new invariants while other threads list them through a barrier Then no thread raised RuntimeError during mixed enforcement and listing And the enforcement record count should be consistent across all threads diff --git a/features/steps/invariant_service_thread_safety_steps.py b/features/steps/invariant_service_thread_safety_steps.py index 56267263d..e0bfa9734 100644 --- a/features/steps/invariant_service_thread_safety_steps.py +++ b/features/steps/invariant_service_thread_safety_steps.py @@ -25,7 +25,9 @@ from cleveragents.domain.models.core.invariant import InvariantScope # ================================================================ -def _create_barrier_and_threads(n: int) -> tuple[threading.Barrier, list[threading.Thread]]: +def _create_barrier_and_threads( + n: int, +) -> tuple[threading.Barrier, list[threading.Thread]]: """Create a barrier for n threads and an empty thread list.""" barrier = threading.Barrier(n) threads: list[threading.Thread] = [] @@ -37,7 +39,7 @@ def _create_barrier_and_threads(n: int) -> tuple[threading.Barrier, list[threadi # ================================================================ -@given("an invariant service") +@given("an invariant service for thread-safety testing") def step_service(context: Context) -> None: """Create a fresh InvariantService with empty concurrency tracking.""" context.service = InvariantService() @@ -70,12 +72,18 @@ def step_service_with_invariants(context: Context, count: int) -> None: source_name="enforcement-test", ) # Store the invariant IDs so enforcement scenarios can reference them - context.enforcement_inv_ids = list(context.concurrent_results) if hasattr(context, "concurrent_results") else [] + context.enforcement_inv_ids = ( + list(context.concurrent_results) + if hasattr(context, "concurrent_results") + else [] + ) context.concurrent_errors = [] context.concurrent_results = {} -@given("an invariant service with {count:d} invariants already enforced by a previous thread") +@given( + "an invariant service with {count:d} invariants already enforced by a previous thread" +) def step_service_with_enforced_invariants(context: Context, count: int) -> None: """Create an InvariantService where enforcement records already exist.""" context.service = InvariantService() @@ -108,6 +116,8 @@ def step_concurrent_add(context: Context, n: int, count: int) -> None: _ = n * count # actual total added by this scenario barrier, threads = _create_barrier_and_threads(n) + context._error_mutex = threading.Lock() + def worker(thread_index: int) -> None: try: barrier.wait(timeout=10) @@ -118,7 +128,7 @@ def step_concurrent_add(context: Context, n: int, count: int) -> None: source_name="concurrent-test", ) except Exception as exc: - with context._error_mutex if hasattr(context, "_error_mutex") else threading.Lock(): + with context._error_mutex: context.concurrent_errors.append(exc) for idx in range(n): @@ -126,10 +136,6 @@ def step_concurrent_add(context: Context, n: int, count: int) -> None: threads.append(t) t.start() - # Also create a mutex for the first call since it may not exist yet - if not hasattr(context, "_error_mutex"): - context._error_mutex = threading.Lock() - for t in threads: t.join(timeout=30) @@ -158,13 +164,17 @@ def step_concurrent_add_count(context: Context, total: int) -> None: @then("all added invariants should be retrievable via get_invariant_snapshot") def step_snapshot_contains_all(context: Context) -> None: - """Confirm get_invariants_snapshot returns consistent data.""" + """Confirm get_invariants_snapshot and get_invariant return consistent data.""" snapshot = context.service.get_invariants_snapshot() listed = context.service.list_invariants() assert len(snapshot) == len(listed), ( f"Snapshot has {len(snapshot)} invariants but list_invariants found " f"{len(listed)}. Inconsistent reads detected." ) + for inv_id in snapshot: + assert context.service.get_invariant(inv_id) is not None, ( + f"get_invariant({inv_id!r}) returned None but id is in snapshot." + ) # ================================================================ @@ -202,21 +212,25 @@ def step_concurrent_list(context: Context) -> None: @then("no thread raised RuntimeError during concurrent reads") def step_no_runtime_error_lists(context: Context) -> None: """Verify no thread hit a RuntimeError during list_invariants().""" - runtime_errors = [e for e in context.concurrent_errors if isinstance(e, RuntimeError)] + runtime_errors = [ + e for e in context.concurrent_errors if isinstance(e, RuntimeError) + ] assert not runtime_errors, ( f"RuntimeError raised during concurrent reads: {runtime_errors}. " "The RLock is missing in list_invariants()." ) -@then("the caller should receive a valid list of at least {min:d} active invariants") -def step_list_produces_valid_result(context: Context, min_: int) -> None: +@then( + "the caller should receive a valid list of at least {min_count:d} active invariants" +) +def step_list_produces_valid_result(context: Context, min_count: int) -> None: """Each thread's list result should be consistent.""" # All threads should have seen the same count (snapshot consistency) for results in context._list_results: - assert len(results) >= min_, ( + assert len(results) >= min_count, ( f"Thread saw only {len(results)} active invariants " - f"(expected at least {min_}). " + f"(expected at least {min_count}). " "A reader may have seen a partially-written dict." ) @@ -279,7 +293,9 @@ def step_concurrent_mixed(context: Context, n: int, m: int) -> None: @then("no thread raised RuntimeError during mixed access") def step_no_runtime_error_mixed(context: Context) -> None: """Verify no RuntimeError was thrown during concurrent add + remove.""" - runtime_errors = [e for e in context.concurrent_errors if isinstance(e, RuntimeError)] + runtime_errors = [ + e for e in context.concurrent_errors if isinstance(e, RuntimeError) + ] assert not runtime_errors, ( f"RuntimeError raised during mixed access: {runtime_errors}. " "The RLock is missing or ineffective for concurrent read+write." @@ -333,7 +349,9 @@ def step_concurrent_enforce(context: Context, n: int) -> None: if not hasattr(context, "_enf_lock"): context._enf_lock = threading.Lock() - threads = [threading.Thread(target=worker, args=(i,), daemon=True) for i in range(n)] + threads = [ + threading.Thread(target=worker, args=(i,), daemon=True) for i in range(n) + ] for t in threads: t.start() for t in threads: @@ -353,7 +371,9 @@ def step_enforce_record_count(context: Context, count: int) -> None: @then("no thread raised RuntimeError during enforcement") def step_no_runtime_error_enforce(context: Context) -> None: """Verify no RuntimeError during concurrent enforce_invariants().""" - runtime_errors = [e for e in context.concurrent_errors if isinstance(e, RuntimeError)] + runtime_errors = [ + e for e in context.concurrent_errors if isinstance(e, RuntimeError) + ] assert not runtime_errors, ( f"RuntimeError raised during enforcement: {runtime_errors}. " "_enforcement_records.extend() is likely unprotected." @@ -414,7 +434,9 @@ def step_concurrent_enforce_list(context: Context, n: int) -> None: @then("no thread raised RuntimeError during mixed enforcement and listing") def step_no_runtime_error_mixed_enforce(context: Context) -> None: """Verify no RuntimeError during concurrent enforce + list.""" - runtime_errors = [e for e in context.concurrent_errors if isinstance(e, RuntimeError)] + runtime_errors = [ + e for e in context.concurrent_errors if isinstance(e, RuntimeError) + ] assert not runtime_errors, ( f"RuntimeError raised during mixed enforcement and listing: {runtime_errors}" ) -- 2.52.0 From af5bb283d584a270f3399c7d992e757a40bc49b9 Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Wed, 17 Jun 2026 18:41:42 -0400 Subject: [PATCH 6/9] chore: re-trigger CI [controller] -- 2.52.0 From 10bea28b91059c3fa8ba8f7162dd5ddf6739ad09 Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Wed, 17 Jun 2026 19:32:32 -0400 Subject: [PATCH 7/9] chore: re-trigger CI [controller] -- 2.52.0 From e2f7142d74115c34c6ad72d561c88d2b95bd8ca3 Mon Sep 17 00:00:00 2001 From: CleverAgents Bot Date: Wed, 17 Jun 2026 22:10:56 -0400 Subject: [PATCH 8/9] ci: rerun invariant thread safety gates -- 2.52.0 From ddca14c06684678b7d0bcd089478f6e032b8230f Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Wed, 17 Jun 2026 22:50:12 -0400 Subject: [PATCH 9/9] chore: re-trigger CI [controller] -- 2.52.0