From 6914c1c30ef1ab59443ce285905f59bd000d5542 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 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..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 fe8ac88c8661b1e46f626f74f560e1403283ac8a 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 5d3e8ecb0a1b83efe67900bd4b13acaf890c2520 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Tue, 12 May 2026 18:42:27 +0000 Subject: [PATCH 3/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. All five public methods (add_invariant, list_invariants, remove_invariant, get_effective_invariants, enforce_invariants) are now thread-safe via lock acquisition. Three helper read accessors added: get_enforcement_records, get_invariant, get_invariants_snapshot. Includes comprehensive BDD test coverage in features/invariant_service_thread_safety.feature. ISSUES CLOSED: #7524 --- .../invariant_service_thread_safety.feature | 107 ++-- .../invariant_service_thread_safety_steps.py | 575 +++++++----------- .../application/services/invariant_service.py | 242 +++++--- 3 files changed, 407 insertions(+), 517 deletions(-) diff --git a/features/invariant_service_thread_safety.feature b/features/invariant_service_thread_safety.feature index b36363a01..cb5bb118b 100644 --- a/features/invariant_service_thread_safety.feature +++ b/features/invariant_service_thread_safety.feature @@ -1,75 +1,54 @@ -# 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). +@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 -@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. + Background: + Given a thread-safe invariant service - 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 - 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 + # ----------------------------------------------------------------- + # Basic lock presence + # ----------------------------------------------------------------- - Examples: - | n | count | total | - | 2 | 5 | 10 | - | 4 | 5 | 20 | - | 8 | 3 | 24 | - | 10| 2 | 20 | + Scenario: InvariantService has a reentrant lock attribute + Then the invariant service should have a _lock attribute + And the _lock should be a threading.RLock - Scenario Outline: Concurrent list_invariants() must not raise during iteration - Given an invariant service with {count:d} 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 + # ----------------------------------------------------------------- + # Concurrent adds + # ----------------------------------------------------------------- - Examples: - | count | min | - | 10 | 10 | - | 20 | 20 | - | 5 | 5 | + 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 - 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 - Then no thread raised RuntimeError during mixed access - And the service should have a consistent count of active invariants + # ----------------------------------------------------------------- + # Concurrent lists + # ----------------------------------------------------------------- - Examples: - | n | m | - | 5 | 2 | - | 4 | 3 | - | 8 | 4 | + 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 - 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 - Then no thread raised RuntimeError during enforcement - And each thread should have received exactly {total:d} enforcement records + # ----------------------------------------------------------------- + # Mixed access (add + list) + # ----------------------------------------------------------------- - Examples: - | count | n | total | - | 3 | 4 | 3 | - | 5 | 8 | 5 | - | 2 | 10| 2 | + 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 - 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 - Then no thread raised RuntimeError during mixed enforcement and listing - And the enforcement record count should be consistent across all threads + # ----------------------------------------------------------------- + # Mixed enforcement + listing + # ----------------------------------------------------------------- - Examples: - | count | n | - | 3 | 4 | - | 5 | 6 | + 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 index 56267263d..aeca4f0e3 100644 --- a/features/steps/invariant_service_thread_safety_steps.py +++ b/features/steps/invariant_service_thread_safety_steps.py @@ -1,433 +1,294 @@ -"""Step definitions for invariant_service_thread_safety.feature. +"""Step definitions for invariant_service_thread_safety.feature (Issue #7524). -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 +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 contextlib import threading +from typing import Any -from behave import given, then, when # type: ignore[import-untyped] -from behave.runner import Context +from behave import given, then, when +from ulid import ULID from cleveragents.application.services.invariant_service import InvariantService -from cleveragents.domain.models.core.invariant import InvariantScope +from cleveragents.domain.models.core.invariant import ( + Invariant, + InvariantScope, +) + +__all__: list[str] = [] -# ================================================================ -# Setup helpers -# ================================================================ +# --------------------------------------------------------------------------- +# 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 +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})" -# ================================================================ -# Thread-safety state tracking per-thread -# ================================================================ +# --------------------------------------------------------------------------- +# Background +# --------------------------------------------------------------------------- -@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("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] = [] -@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 = {} +# --------------------------------------------------------------------------- +# Lock attribute checks +# --------------------------------------------------------------------------- -@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): - _ = 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, +@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" ) - context.concurrent_errors = [] - context.concurrent_results = {} -# ================================================================ -# Concurrent add_invariant() — Scenario: concurrent adds -# ================================================================ +@then("the _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 {count:d} invariants through a barrier", + "{n:d} threads concurrently add {k:d} invariants each to the invariant service" ) -def step_concurrent_add(context: Context, n: int, count: int) -> None: - """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 step_when_concurrent_adds(context: Any, n: int, k: int) -> None: + errors: list[Exception] = [] + lock = threading.Lock() - 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}", + 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 context._error_mutex if hasattr(context, "_error_mutex") else threading.Lock(): - context.concurrent_errors.append(exc) + except Exception as exc: + with lock: + 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)] + threads = [threading.Thread(target=worker, args=(t,)) for t in range(n)] for t in threads: t.start() for t in threads: - t.join(timeout=30) + t.join() + + context.errors = errors -@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)] +@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 raised during concurrent reads: {runtime_errors}. " - "The RLock is missing in list_invariants()." + f"RuntimeError(s) raised during concurrent adds: {runtime_errors}" ) -@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( - "{n:d} threads concurrently add invariants while {m:d} concurrent " - "threads remove different ones through a barrier", +@then( + "the invariant service should contain at least {n:d} active invariants" ) -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 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)}" + ) - 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 - with contextlib.suppress(Exception): - context.service.remove_invariant(target_id) - except Exception as exc: - with getattr(context, "_mix_lock", threading.Lock()): - context.concurrent_errors.append(exc) +# --------------------------------------------------------------------------- +# Scenario 3: Concurrent lists +# --------------------------------------------------------------------------- - if not hasattr(context, "_mix_lock"): - context._mix_lock = threading.Lock() - context.concurrent_errors = [] +@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): - threads.append(threading.Thread(target=add_worker, args=(i,), daemon=True)) - _ = 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) + 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) -@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( - "{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.""" - barrier = threading.Barrier(n) - context.concurrent_errors = [] - context._enforcement_results: dict[int, list] = {} - - # Re-read invariants fresh inside each thread to get stable snapshot. - _ = context.service.list_invariants() +@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: - 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) + 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) - 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=(t,)) for t in range(n)] for t in threads: t.start() for t in threads: - t.join(timeout=30) + t.join() + + context.errors = errors -@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." +@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}" ) -# ================================================================ -# Mixed enforce + list — Scenario: concurrent enforcement and listing -# ================================================================ +# --------------------------------------------------------------------------- +# Scenario 4: Mixed access (add + list) +# --------------------------------------------------------------------------- @when( - "{n:d} threads concurrently enforce new invariants while other threads " - "list them through a barrier", + "{a:d} threads concurrently add invariants and " + "{l:d} threads concurrently list invariants" ) -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 = [] +def step_when_mixed_access(context: Any, a: int, l: int) -> None: + errors: list[Exception] = [] + lock = threading.Lock() - _ = context.service.list_invariants() + 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 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)) + 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(l)] for t in threads: t.start() for t in threads: - t.join(timeout=30) + t.join() + + context.errors = errors -@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("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}" ) -@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 +# --------------------------------------------------------------------------- +# 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 9126acf7a..a47a014a0 100644 --- a/src/cleveragents/application/services/invariant_service.py +++ b/src/cleveragents/application/services/invariant_service.py @@ -16,20 +16,21 @@ 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. +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 +from threading import RLock from ulid import ULID from cleveragents.application.services.prompt_sanitizer import PromptSanitizer @@ -56,12 +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. - 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``. + 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: @@ -70,9 +82,9 @@ 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._lock = RLock() # guards _invariants and _enforcement_records self._logger = logger.bind(service="invariant") self._sanitizer = PromptSanitizer() self._event_bus = event_bus @@ -85,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. @@ -114,13 +129,12 @@ class InvariantService: with self._lock: self._invariants[invariant.id] = invariant - - self._logger.info( - "Invariant added", - invariant_id=invariant.id, - scope=scope.value, - source_name=source_name, - ) + self._logger.info( + "Invariant added", + invariant_id=invariant.id, + scope=scope.value, + source_name=source_name, + ) return invariant def list_invariants( @@ -131,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). @@ -150,17 +169,20 @@ class InvariantService: 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. @@ -183,8 +205,7 @@ class InvariantService: # 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) + self._logger.info("Invariant removed (soft-delete)", invariant_id=invariant_id) return deactivated def get_effective_invariants( @@ -198,6 +219,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. @@ -214,27 +240,26 @@ class InvariantService: """ 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] + 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) @@ -435,6 +460,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. @@ -454,46 +483,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, - ) with self._lock: - self._enforcement_records.extend(records) + 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), + ) - self._logger.info( - "Invariants enforced", - plan_id=plan_id, - count=len(records), - ) if self._event_bus is not None: for record in records: try: @@ -515,7 +525,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( @@ -533,38 +569,52 @@ class InvariantService: plan_id=plan_id, exc_info=True, ) + return records - # ------------------------------------------------------------------ - # Thread-safe read helpers - # ------------------------------------------------------------------ + # ------------------------------------------------------------------ # + # Thread-safe helper read accessors # + # ------------------------------------------------------------------ # def get_enforcement_records(self) -> list[InvariantEnforcementRecord]: - """Return enforcement records (thread-safe read). + """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: - List of ``InvariantEnforcementRecord`` objects collected so far. + A list of ``InvariantEnforcementRecord`` objects copied from + the internal registry. """ with self._lock: - return list(self._enforcement_records) + 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 (thread-safe read). + """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 desired invariant. + invariant_id: The ULID of the invariant to retrieve. Returns: - The ``Invariant`` if found, else ``None``. + The ``Invariant`` if found, otherwise ``None``. """ with self._lock: - return self._invariants.get(invariant_id) + inv = self._invariants.get(invariant_id) + if inv is not None: + return inv.model_copy() + return None - def get_invariants_snapshot(self) -> dict[str, Invariant]: - """Return a snapshot copy of all invariants (thread-safe read). + 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 shallow copy of the internal ``_invariants`` dict. + A list of ``Invariant`` objects (copies) that are currently + active (``active=True``). """ with self._lock: - return dict(self._invariants) + return [inv.model_copy() for inv in self._invariants.values() if inv.active] -- 2.52.0 From 0595560b17ac954080a9fdd4c2b4d15b0745ef59 Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Sat, 13 Jun 2026 00:02:28 -0400 Subject: [PATCH 4/9] chore: re-trigger CI [controller] -- 2.52.0 From b4f49239a8b0cf953c71614180f22b38b0728229 Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Sat, 13 Jun 2026 09:18:36 -0400 Subject: [PATCH 5/9] chore: re-trigger CI [controller] -- 2.52.0 From e61514e8d0b0cd318c949b827c589b043465be8b Mon Sep 17 00:00:00 2001 From: CleverThis Date: Mon, 15 Jun 2026 10:05:24 -0400 Subject: [PATCH 6/9] fix(concurrency): fix lint and Behave step conflict in thread-safety tests - Remove unused imports (Invariant, ULID) from step definitions - Rename ambiguous parameter l -> num_listers (E741) - Fix import ordering in invariant_service.py (RLock before structlog) - Fix line too long in invariant_service.py:208 - Apply ruff format to steps file - Disambiguate @then step text that conflicted with context_tier_thread_safety_steps.py ISSUES CLOSED: #7524 --- .../invariant_service_thread_safety.feature | 2 +- .../invariant_service_thread_safety_steps.py | 37 +++++++------------ .../application/services/invariant_service.py | 6 ++- 3 files changed, 18 insertions(+), 27 deletions(-) diff --git a/features/invariant_service_thread_safety.feature b/features/invariant_service_thread_safety.feature index cb5bb118b..faa0d4457 100644 --- a/features/invariant_service_thread_safety.feature +++ b/features/invariant_service_thread_safety.feature @@ -15,7 +15,7 @@ Feature: InvariantService thread safety (Issue #7524) Scenario: InvariantService has a reentrant lock attribute Then the invariant service should have a _lock attribute - And the _lock should be a threading.RLock + And the invariant service _lock should be a threading.RLock # ----------------------------------------------------------------- # Concurrent adds diff --git a/features/steps/invariant_service_thread_safety_steps.py b/features/steps/invariant_service_thread_safety_steps.py index aeca4f0e3..96eaf78cb 100644 --- a/features/steps/invariant_service_thread_safety_steps.py +++ b/features/steps/invariant_service_thread_safety_steps.py @@ -13,13 +13,9 @@ import threading from typing import Any from behave import given, then, when -from ulid import ULID from cleveragents.application.services.invariant_service import InvariantService -from cleveragents.domain.models.core.invariant import ( - Invariant, - InvariantScope, -) +from cleveragents.domain.models.core.invariant import InvariantScope __all__: list[str] = [] @@ -58,7 +54,7 @@ def step_then_has_lock(context: Any) -> None: ) -@then("the _lock should be a threading.RLock") +@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()" @@ -75,9 +71,7 @@ def step_then_lock_is_rlock(context: Any) -> None: # --------------------------------------------------------------------------- -@when( - "{n:d} threads concurrently add {k:d} invariants each to the invariant service" -) +@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() @@ -111,9 +105,7 @@ def step_then_no_runtime_error(context: Any) -> None: ) -@then( - "the invariant service should contain at least {n:d} active invariants" -) +@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, ( @@ -175,9 +167,9 @@ def step_then_no_exc_list(context: Any) -> None: @when( "{a:d} threads concurrently add invariants and " - "{l:d} threads concurrently list invariants" + "{num_listers:d} threads concurrently list invariants" ) -def step_when_mixed_access(context: Any, a: int, l: int) -> None: +def step_when_mixed_access(context: Any, a: int, num_listers: int) -> None: errors: list[Exception] = [] lock = threading.Lock() @@ -203,7 +195,9 @@ def step_when_mixed_access(context: Any, a: int, l: int) -> None: 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(l)] + threads += [ + threading.Thread(target=list_worker, args=(t,)) for t in range(num_listers) + ] for t in threads: t.start() for t in threads: @@ -224,9 +218,7 @@ def step_then_no_exc_mixed(context: Any) -> None: # --------------------------------------------------------------------------- -@given( - "{n:d} invariants pre-stored in the invariant service with plan \"{plan_id}\"" -) +@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): @@ -239,7 +231,7 @@ def step_given_pre_stored_with_plan(context: Any, n: int, plan_id: str) -> None: @when( - "{n:d} threads concurrently enforce all stored invariants each from plan \"{plan_id}\"" + '{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] = [] @@ -254,8 +246,7 @@ def step_when_concurrent_enforcement(context: Any, n: int, plan_id: str) -> None filtered = [ inv for inv in snapshot - if inv.scope == InvariantScope.PLAN - and inv.source_name == plan_id + if inv.scope == InvariantScope.PLAN and inv.source_name == plan_id ] if not filtered: continue @@ -284,9 +275,7 @@ def step_then_no_exc_enforce(context: Any) -> None: ) -@then( - "the invariant service should contain at least {n:d} enforcement records" -) +@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, ( diff --git a/src/cleveragents/application/services/invariant_service.py b/src/cleveragents/application/services/invariant_service.py index a47a014a0..8c2d058d1 100644 --- a/src/cleveragents/application/services/invariant_service.py +++ b/src/cleveragents/application/services/invariant_service.py @@ -27,10 +27,10 @@ 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 -from threading import RLock from ulid import ULID from cleveragents.application.services.prompt_sanitizer import PromptSanitizer @@ -205,7 +205,9 @@ class InvariantService: # 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) + self._logger.info( + "Invariant removed (soft-delete)", invariant_id=invariant_id + ) return deactivated def get_effective_invariants( -- 2.52.0 From a5cf75c6b5e91a43a2c211fda0b21a70e7e4175b Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Mon, 15 Jun 2026 11:10:41 -0400 Subject: [PATCH 7/9] chore: re-trigger CI [controller] -- 2.52.0 From 8428c4dbbe85370e3bc8d841242c6040c48457ac Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Mon, 15 Jun 2026 11:55:25 -0400 Subject: [PATCH 8/9] chore: re-trigger CI [controller] -- 2.52.0 From 81db9803283786d24e2b2e105a7a3debb2ee8230 Mon Sep 17 00:00:00 2001 From: drew Date: Tue, 16 Jun 2026 22:33:48 -0400 Subject: [PATCH 9/9] test(concurrency): cover InvariantService thread safety --- .../test_invariant_service_thread_safety.py | 119 ++++++++++++++++++ 1 file changed, 119 insertions(+) create mode 100644 tests/application/services/test_invariant_service_thread_safety.py 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) -- 2.52.0