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(