From 0f7542925307918f44786d17ba811ec757d9a9d4 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Mon, 13 Apr 2026 07:29:00 +0000 Subject: [PATCH 1/2] fix(concurrency): add thread safety to ContextTierService Implemented thread-safety improvements for ContextTierService by introducing a re-entrant lock and guarding all critical sections with self._lock. This prevents RuntimeError: dictionary changed size during iteration under concurrent plan execution. - Added threading.RLock to ContextTierService.__init__ as self._lock - Wrapped all public methods (store, get, promote, demote, evict_lru, get_metrics, get_all_fragments, get_hot_fragments, get_for_actor, get_scoped_view) with with self._lock: - Added _lock: threading.RLock type stub to TierRuntimeMixin and ScopedTierMixin - Wrapped enforce_staleness in TierRuntimeMixin with self._lock - Wrapped get_scoped_by_resource and get_scoped_metrics in ScopedTierMixin with self._lock - Extracted settings helpers to new context_tier_settings.py to keep context_tiers.py under 500 lines - Added BDD feature file context_tier_thread_safety.feature with 10 thread-safety scenarios - Added step definitions context_tier_thread_safety_steps.py - Updated CHANGELOG.md with fix entry ISSUES CLOSED: #7547 --- CHANGELOG.md | 15 + features/context_tier_thread_safety.feature | 84 +++ .../steps/context_tier_thread_safety_steps.py | 461 +++++++++++++++ .../services/context_tier_settings.py | 103 ++++ .../application/services/context_tiers.py | 550 ++++++++---------- .../application/services/scoped_tiers.py | 67 ++- .../application/services/tier_runtime.py | 77 +-- 7 files changed, 967 insertions(+), 390 deletions(-) create mode 100644 features/context_tier_thread_safety.feature create mode 100644 features/steps/context_tier_thread_safety_steps.py create mode 100644 src/cleveragents/application/services/context_tier_settings.py diff --git a/CHANGELOG.md b/CHANGELOG.md index a2a1900d8..91bafb3f6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -177,6 +177,21 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). sequences. The `color` format is now routed to `format_output_session` which uses the `ColorMaterializer` to emit proper ANSI-coloured output. `--format plain` and all other formats remain unaffected. + +- **ContextTierService Thread Safety** (#7547): Added `threading.RLock` to + `ContextTierService` to prevent `RuntimeError: dictionary changed size during + iteration` and data corruption under concurrent plan execution. All public + methods (`store`, `get`, `promote`, `demote`, `evict_lru`, `enforce_staleness`, + `get_metrics`, `get_all_fragments`, `get_hot_fragments`, `get_for_actor`, + `get_scoped_view`, `get_scoped_by_resource`, `get_scoped_metrics`) now acquire + the reentrant lock before accessing the hot/warm/cold tier dicts. The service + was previously documented as single-threaded but registered as a DI Singleton, + causing potential data corruption when parallel subplans shared the same + instance. The `TierRuntimeMixin.enforce_staleness()` and + `ScopedTierMixin.get_scoped_by_resource()` / `get_scoped_metrics()` methods + are also protected. The DI container registration as `providers.Singleton` + is now correct and safe. + - **Validation Gate Empty-Run Guard** (#7508): Fixed `ApplyValidationSummary.all_required_passed` returning `True` when zero validations were run, silently bypassing the apply gate. The property now returns `False` when the validation result set is empty (`is_empty` is `True`), ensuring diff --git a/features/context_tier_thread_safety.feature b/features/context_tier_thread_safety.feature new file mode 100644 index 000000000..fb393479a --- /dev/null +++ b/features/context_tier_thread_safety.feature @@ -0,0 +1,84 @@ +@mock_only +Feature: ContextTierService thread safety (#7547) + As an ACMS developer + I want ContextTierService 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 the hot/warm/cold tier dicts + + Background: + Given a thread-safe context tier service + + # ----------------------------------------------------------------- + # Basic lock presence + # ----------------------------------------------------------------- + + Scenario: ContextTierService has a reentrant lock attribute + Then the tier service should have a _lock attribute + And the _lock should be a threading.RLock + + # ----------------------------------------------------------------- + # Concurrent store operations + # ----------------------------------------------------------------- + + Scenario: Concurrent store from multiple threads does not raise RuntimeError + When 10 threads concurrently store 5 fragments each into the tier service + Then no RuntimeError should have been raised during concurrent store + And the tier service should contain at least 1 fragment + + Scenario: Concurrent store and get from multiple threads is data-race free + Given 20 fragments pre-stored in the tier service + When 5 threads concurrently store fragments and 5 threads concurrently get fragments + Then no exception should have been raised during concurrent store and get + + # ----------------------------------------------------------------- + # Concurrent promote / demote + # ----------------------------------------------------------------- + + Scenario: Concurrent promote from multiple threads does not corrupt tier state + Given 10 fragments stored in the cold tier + When 5 threads concurrently promote all cold fragments + Then no exception should have been raised during concurrent promote + And no fragment should appear in more than one tier + + Scenario: Concurrent demote from multiple threads does not corrupt tier state + Given 10 fragments stored in the hot tier + When 5 threads concurrently demote all hot fragments + Then no exception should have been raised during concurrent demote + And no fragment should appear in more than one tier + + # ----------------------------------------------------------------- + # Concurrent evict_lru + # ----------------------------------------------------------------- + + Scenario: Concurrent evict_lru does not raise RuntimeError + Given 20 fragments stored in the hot tier + When 4 threads concurrently evict 1 LRU fragment from the hot tier + Then no exception should have been raised during concurrent evict + + # ----------------------------------------------------------------- + # Concurrent enforce_staleness + # ----------------------------------------------------------------- + + Scenario: Concurrent enforce_staleness does not raise RuntimeError + Given 10 fragments stored in the hot tier with stale timestamps + When 4 threads concurrently invoke enforce_staleness + Then no exception should have been raised during concurrent staleness enforcement + + # ----------------------------------------------------------------- + # Concurrent get_metrics + # ----------------------------------------------------------------- + + Scenario: Concurrent get_metrics does not raise RuntimeError + Given 10 fragments stored in the hot tier + When 8 threads concurrently call get_metrics while 4 threads store fragments + Then no exception should have been raised during concurrent metrics access + + # ----------------------------------------------------------------- + # Singleton safety: same instance shared across threads + # ----------------------------------------------------------------- + + Scenario: Shared singleton instance survives concurrent access from 20 threads + When 20 threads each store 3 fragments and immediately get them back + Then no exception should have been raised during concurrent singleton access + And the tier service metrics should reflect a non-negative fragment count diff --git a/features/steps/context_tier_thread_safety_steps.py b/features/steps/context_tier_thread_safety_steps.py new file mode 100644 index 000000000..29dcd3f2e --- /dev/null +++ b/features/steps/context_tier_thread_safety_steps.py @@ -0,0 +1,461 @@ +"""Step definitions for ContextTierService thread-safety tests (#7547). + +Verifies that all public methods of ContextTierService are safe for +concurrent access from multiple threads, preventing: +- RuntimeError: dictionary changed size during iteration +- Data corruption in hot/warm/cold tier dicts +- Lost updates or phantom reads under concurrent plan execution +""" + +from __future__ import annotations + +import threading +from datetime import UTC, datetime, timedelta +from typing import Any + +from behave import given, then, when + +from cleveragents.application.services.context_tiers import ContextTierService +from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment + +__all__: list[str] = [] + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_THREAD_COUNT_DEFAULT = 10 + + +def _make_fragment( + fid: str, + tier: ContextTier = ContextTier.HOT, + token_count: int = 10, + stale: bool = False, +) -> TieredFragment: + """Create a TieredFragment for testing.""" + frag = TieredFragment( + fragment_id=fid, + content=f"content-{fid}", + tier=tier, + token_count=token_count, + ) + if stale: + frag.last_accessed = datetime.now(tz=UTC) - timedelta(hours=48) + return frag + + +# --------------------------------------------------------------------------- +# Background +# --------------------------------------------------------------------------- + + +@given("a thread-safe context tier service") +def step_given_thread_safe_service(context: Any) -> None: + context.tier_service = ContextTierService() + context.errors = [] + + +# --------------------------------------------------------------------------- +# Lock attribute checks +# --------------------------------------------------------------------------- + + +@then("the tier service should have a _lock attribute") +def step_then_has_lock(context: Any) -> None: + assert hasattr(context.tier_service, "_lock"), ( + "ContextTierService must have a _lock attribute" + ) + + +@then("the _lock should be a threading.RLock") +def step_then_lock_is_rlock(context: Any) -> None: + # threading.RLock() returns an instance of _RLock (internal type). + # We verify it has the acquire/release interface of a reentrant lock. + lock = context.tier_service._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() + + +# --------------------------------------------------------------------------- +# Concurrent store +# --------------------------------------------------------------------------- + + +@when("{n:d} threads concurrently store {k:d} fragments each into the tier service") +def step_when_concurrent_store(context: Any, n: int, k: int) -> None: + errors: list[Exception] = [] + lock = threading.Lock() + + def worker(thread_idx: int) -> None: + for i in range(k): + fid = f"ts-store-t{thread_idx}-f{i}" + frag = _make_fragment(fid, tier=ContextTier.HOT, token_count=5) + try: + context.tier_service.store(frag) + except Exception as exc: + with lock: + errors.append(exc) + + threads = [threading.Thread(target=worker, args=(t,)) for t in range(n)] + for t in threads: + t.start() + for t in threads: + t.join() + + context.errors = errors + + +@then("no RuntimeError should have been raised during concurrent store") +def step_then_no_runtime_error_store(context: Any) -> None: + runtime_errors = [e for e in context.errors if isinstance(e, RuntimeError)] + assert not runtime_errors, ( + f"RuntimeError(s) raised during concurrent store: {runtime_errors}" + ) + + +@then("the tier service should contain at least {n:d} fragment") +def step_then_contains_at_least(context: Any, n: int) -> None: + metrics = context.tier_service.get_metrics() + total = metrics.hot_count + metrics.warm_count + metrics.cold_count + assert total >= n, f"Expected at least {n} fragments, got {total}" + + +# --------------------------------------------------------------------------- +# Concurrent store + get +# --------------------------------------------------------------------------- + + +@given("{n:d} fragments pre-stored in the tier service") +def step_given_pre_stored(context: Any, n: int) -> None: + for i in range(n): + frag = _make_fragment(f"pre-{i}", tier=ContextTier.HOT, token_count=5) + context.tier_service.store(frag) + + +@when( + "{s:d} threads concurrently store fragments and " + "{g:d} threads concurrently get fragments" +) +def step_when_concurrent_store_and_get(context: Any, s: int, g: int) -> None: + errors: list[Exception] = [] + lock = threading.Lock() + + def store_worker(thread_idx: int) -> None: + for i in range(5): + fid = f"sg-store-t{thread_idx}-f{i}" + frag = _make_fragment(fid, tier=ContextTier.HOT, token_count=5) + try: + context.tier_service.store(frag) + except Exception as exc: + with lock: + errors.append(exc) + + def get_worker(thread_idx: int) -> None: + for i in range(20): + fid = f"pre-{i % 20}" + try: + context.tier_service.get(fid) + except Exception as exc: + with lock: + errors.append(exc) + + threads = [threading.Thread(target=store_worker, args=(t,)) for t in range(s)] + threads += [threading.Thread(target=get_worker, args=(t,)) for t in range(g)] + for t in threads: + t.start() + for t in threads: + t.join() + + context.errors = errors + + +@then("no exception should have been raised during concurrent store and get") +def step_then_no_exc_store_get(context: Any) -> None: + assert not context.errors, ( + f"Exception(s) raised during concurrent store+get: {context.errors}" + ) + + +# --------------------------------------------------------------------------- +# Concurrent promote +# --------------------------------------------------------------------------- + + +@given("{n:d} fragments stored in the cold tier") +def step_given_cold_fragments(context: Any, n: int) -> None: + context.fragment_ids = [] + for i in range(n): + fid = f"cold-{i}" + frag = _make_fragment(fid, tier=ContextTier.COLD, token_count=5) + context.tier_service.store(frag) + context.fragment_ids.append(fid) + + +@when("{n:d} threads concurrently promote all cold fragments") +def step_when_concurrent_promote(context: Any, n: int) -> None: + errors: list[Exception] = [] + lock = threading.Lock() + fids = list(context.fragment_ids) + + def worker() -> None: + for fid in fids: + try: + context.tier_service.promote(fid) + except Exception as exc: + with lock: + errors.append(exc) + + threads = [threading.Thread(target=worker) for _ 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 promote") +def step_then_no_exc_promote(context: Any) -> None: + assert not context.errors, ( + f"Exception(s) raised during concurrent promote: {context.errors}" + ) + + +@then("no fragment should appear in more than one tier") +def step_then_no_duplicate_tier(context: Any) -> None: + svc = context.tier_service + # Access private stores directly to verify no duplicates + hot_ids = set(svc._hot.keys()) + warm_ids = set(svc._warm.keys()) + cold_ids = set(svc._cold.keys()) + + hot_warm = hot_ids & warm_ids + hot_cold = hot_ids & cold_ids + warm_cold = warm_ids & cold_ids + + assert not hot_warm, f"Fragments in both hot and warm: {hot_warm}" + assert not hot_cold, f"Fragments in both hot and cold: {hot_cold}" + assert not warm_cold, f"Fragments in both warm and cold: {warm_cold}" + + +# --------------------------------------------------------------------------- +# Concurrent demote +# --------------------------------------------------------------------------- + + +@given("{n:d} fragments stored in the hot tier") +def step_given_hot_fragments(context: Any, n: int) -> None: + context.fragment_ids = [] + for i in range(n): + fid = f"hot-{i}" + frag = _make_fragment(fid, tier=ContextTier.HOT, token_count=5) + context.tier_service.store(frag) + context.fragment_ids.append(fid) + + +@when("{n:d} threads concurrently demote all hot fragments") +def step_when_concurrent_demote(context: Any, n: int) -> None: + errors: list[Exception] = [] + lock = threading.Lock() + fids = list(context.fragment_ids) + + def worker() -> None: + for fid in fids: + try: + context.tier_service.demote(fid) + except Exception as exc: + with lock: + errors.append(exc) + + threads = [threading.Thread(target=worker) for _ 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 demote") +def step_then_no_exc_demote(context: Any) -> None: + assert not context.errors, ( + f"Exception(s) raised during concurrent demote: {context.errors}" + ) + + +# --------------------------------------------------------------------------- +# Concurrent evict_lru +# --------------------------------------------------------------------------- + + +@when("{n:d} threads concurrently evict {k:d} LRU fragment from the hot tier") +def step_when_concurrent_evict(context: Any, n: int, k: int) -> None: + errors: list[Exception] = [] + lock = threading.Lock() + + def worker() -> None: + try: + context.tier_service.evict_lru(ContextTier.HOT, k) + except Exception as exc: + with lock: + errors.append(exc) + + threads = [threading.Thread(target=worker) for _ 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 evict") +def step_then_no_exc_evict(context: Any) -> None: + assert not context.errors, ( + f"Exception(s) raised during concurrent evict: {context.errors}" + ) + + +# --------------------------------------------------------------------------- +# Concurrent enforce_staleness +# --------------------------------------------------------------------------- + + +@given("{n:d} fragments stored in the hot tier with stale timestamps") +def step_given_stale_hot_fragments(context: Any, n: int) -> None: + context.fragment_ids = [] + for i in range(n): + fid = f"stale-hot-{i}" + frag = _make_fragment(fid, tier=ContextTier.HOT, token_count=5, stale=True) + context.tier_service.store(frag) + # Manually set stale timestamp after storing + if fid in context.tier_service._hot: + context.tier_service._hot[fid].last_accessed = datetime.now( + tz=UTC + ) - timedelta(hours=48) + context.fragment_ids.append(fid) + + +@when("{n:d} threads concurrently invoke enforce_staleness") +def step_when_concurrent_staleness(context: Any, n: int) -> None: + errors: list[Exception] = [] + lock = threading.Lock() + + def worker() -> None: + try: + context.tier_service.enforce_staleness() + except Exception as exc: + with lock: + errors.append(exc) + + threads = [threading.Thread(target=worker) for _ 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 staleness enforcement") +def step_then_no_exc_staleness(context: Any) -> None: + assert not context.errors, ( + f"Exception(s) raised during concurrent enforce_staleness: {context.errors}" + ) + + +# --------------------------------------------------------------------------- +# Concurrent get_metrics +# --------------------------------------------------------------------------- + + +@when("{n:d} threads concurrently call get_metrics while {m:d} threads store fragments") +def step_when_concurrent_metrics(context: Any, n: int, m: int) -> None: + errors: list[Exception] = [] + lock = threading.Lock() + + def metrics_worker(thread_idx: int) -> None: + for _ in range(10): + try: + context.tier_service.get_metrics() + except Exception as exc: + with lock: + errors.append(exc) + + def store_worker(thread_idx: int) -> None: + for i in range(5): + fid = f"metrics-t{thread_idx}-f{i}" + frag = _make_fragment(fid, tier=ContextTier.HOT, token_count=5) + try: + context.tier_service.store(frag) + except Exception as exc: + with lock: + errors.append(exc) + + threads = [threading.Thread(target=metrics_worker, args=(t,)) for t in range(n)] + threads += [threading.Thread(target=store_worker, args=(t,)) for t in range(m)] + for t in threads: + t.start() + for t in threads: + t.join() + + context.errors = errors + + +@then("no exception should have been raised during concurrent metrics access") +def step_then_no_exc_metrics(context: Any) -> None: + assert not context.errors, ( + f"Exception(s) raised during concurrent get_metrics: {context.errors}" + ) + + +# --------------------------------------------------------------------------- +# Singleton concurrent access +# --------------------------------------------------------------------------- + + +@when("{n:d} threads each store {k:d} fragments and immediately get them back") +def step_when_concurrent_singleton(context: Any, n: int, k: int) -> None: + errors: list[Exception] = [] + lock = threading.Lock() + + def worker(thread_idx: int) -> None: + for i in range(k): + fid = f"singleton-t{thread_idx}-f{i}" + frag = _make_fragment(fid, tier=ContextTier.HOT, token_count=5) + try: + context.tier_service.store(frag) + result = context.tier_service.get(fid) + # result may be None if another thread evicted it; that's OK + _ = result + 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 singleton access") +def step_then_no_exc_singleton(context: Any) -> None: + assert not context.errors, ( + f"Exception(s) raised during concurrent singleton access: {context.errors}" + ) + + +@then("the tier service metrics should reflect a non-negative fragment count") +def step_then_non_negative_count(context: Any) -> None: + metrics = context.tier_service.get_metrics() + total = metrics.hot_count + metrics.warm_count + metrics.cold_count + assert total >= 0, f"Negative fragment count: {total}" diff --git a/src/cleveragents/application/services/context_tier_settings.py b/src/cleveragents/application/services/context_tier_settings.py new file mode 100644 index 000000000..d52f74831 --- /dev/null +++ b/src/cleveragents/application/services/context_tier_settings.py @@ -0,0 +1,103 @@ +"""Settings helpers for ContextTierService. + +Extracted from ``context_tiers.py`` to keep that module under the +500-line guideline. Provides budget and runtime-policy helpers that +read from application settings (or fall back to defaults). + +Based on ``docs/specification.md`` ACMS tier sections and issue #7547. +""" + +from __future__ import annotations + +from datetime import timedelta + +from cleveragents.config.settings import Settings +from cleveragents.domain.models.acms.tiers import TierBudget + +# --------------------------------------------------------------------------- +# Default budget when settings are not provided +# --------------------------------------------------------------------------- + +DEFAULT_MAX_TOKENS_HOT = 8000 +DEFAULT_MAX_DECISIONS_WARM = 500 +DEFAULT_MAX_DECISIONS_COLD = 5000 + +# --------------------------------------------------------------------------- +# Default runtime policy values +# --------------------------------------------------------------------------- + +DEFAULT_PROMOTION_THRESHOLD = 5 +DEFAULT_HOT_TTL_HOURS = 24 +DEFAULT_WARM_TTL_HOURS = 24 + + +def budget_from_settings(settings: Settings | None) -> TierBudget: + """Build a ``TierBudget`` from application settings (or defaults).""" + if settings is None: + return TierBudget() + return TierBudget( + max_tokens_hot=getattr( + settings, "context_max_tokens_hot", DEFAULT_MAX_TOKENS_HOT + ), + max_decisions_warm=getattr( + settings, "context_max_decisions_warm", DEFAULT_MAX_DECISIONS_WARM + ), + max_decisions_cold=getattr( + settings, "context_max_decisions_cold", DEFAULT_MAX_DECISIONS_COLD + ), + ) + + +def promotion_threshold_from_settings(settings: Settings | None) -> int: + """Read promotion threshold from settings (or use default).""" + if settings is None: + return DEFAULT_PROMOTION_THRESHOLD + return int( + getattr( + settings, + "context_tier_promotion_threshold", + DEFAULT_PROMOTION_THRESHOLD, + ), + ) + + +def hot_ttl_from_settings(settings: Settings | None) -> timedelta: + """Read hot-tier TTL from settings (or use default).""" + if settings is None: + return timedelta(hours=DEFAULT_HOT_TTL_HOURS) + hours = int( + getattr( + settings, + "context_tier_hot_ttl_hours", + DEFAULT_HOT_TTL_HOURS, + ), + ) + return timedelta(hours=hours) + + +def warm_ttl_from_settings(settings: Settings | None) -> timedelta: + """Read warm-tier TTL from settings (or use default).""" + if settings is None: + return timedelta(hours=DEFAULT_WARM_TTL_HOURS) + hours = int( + getattr( + settings, + "context_tier_warm_ttl_hours", + DEFAULT_WARM_TTL_HOURS, + ), + ) + return timedelta(hours=hours) + + +__all__ = [ + "DEFAULT_HOT_TTL_HOURS", + "DEFAULT_MAX_DECISIONS_COLD", + "DEFAULT_MAX_DECISIONS_WARM", + "DEFAULT_MAX_TOKENS_HOT", + "DEFAULT_PROMOTION_THRESHOLD", + "DEFAULT_WARM_TTL_HOURS", + "budget_from_settings", + "hot_ttl_from_settings", + "promotion_threshold_from_settings", + "warm_ttl_from_settings", +] diff --git a/src/cleveragents/application/services/context_tiers.py b/src/cleveragents/application/services/context_tiers.py index 03d0743cb..cf71c92cc 100644 --- a/src/cleveragents/application/services/context_tiers.py +++ b/src/cleveragents/application/services/context_tiers.py @@ -8,17 +8,28 @@ eviction, event emission) is provided by ``TierRuntimeMixin`` in Scope enforcement methods live in ``scoped_tiers.py`` and are mixed in via ``ScopedTierMixin``. +Settings helpers (budget, TTL, promotion threshold) live in +``context_tier_settings.py``. + Based on ``docs/specification.md`` ACMS tier sections, issue #208, -issue #193 (scoped backend view filtering), and issue #821 -(runtime promotion/demotion/eviction). +issue #193 (scoped backend view filtering), issue #821 +(runtime promotion/demotion/eviction), and issue #7547 +(thread-safety: RLock protection for concurrent plan execution). """ from __future__ import annotations -from datetime import UTC, datetime, timedelta +import threading +from datetime import UTC, datetime import structlog +from cleveragents.application.services.context_tier_settings import ( + budget_from_settings, + hot_ttl_from_settings, + promotion_threshold_from_settings, + warm_ttl_from_settings, +) from cleveragents.application.services.scoped_tiers import ScopedTierMixin from cleveragents.application.services.tier_runtime import TierRuntimeMixin from cleveragents.config.settings import Settings @@ -38,22 +49,6 @@ from cleveragents.infrastructure.events.types import EventType logger = structlog.get_logger(__name__) -# --------------------------------------------------------------------------- -# Default budget when settings are not provided -# --------------------------------------------------------------------------- - -_DEFAULT_MAX_TOKENS_HOT = 8000 -_DEFAULT_MAX_DECISIONS_WARM = 500 -_DEFAULT_MAX_DECISIONS_COLD = 5000 - -# --------------------------------------------------------------------------- -# Default runtime policy values -# --------------------------------------------------------------------------- - -_DEFAULT_PROMOTION_THRESHOLD = 5 -_DEFAULT_HOT_TTL_HOURS = 24 -_DEFAULT_WARM_TTL_HOURS = 24 - # Maximum content length kept after cold-tier summarisation _COLD_SUMMARY_MAX_CHARS = 200 @@ -61,37 +56,20 @@ _COLD_SUMMARY_MAX_CHARS = 200 class ContextTierService(TierRuntimeMixin, ScopedTierMixin): """Manage context fragments across hot/warm/cold tiers. - Runtime behaviour (issue #821): - - * **Auto-promotion on access**: When ``get()`` is called and the - fragment's ``access_count`` reaches the promotion threshold, the - fragment is automatically promoted one tier up. - * **Staleness enforcement**: ``enforce_staleness()`` inspects every - fragment's ``last_accessed`` timestamp and demotes stale fragments - (hot → warm after ``hot_ttl``, warm → cold after ``warm_ttl``). - * **Budget enforcement on store**: ``store()`` enforces - ``TierBudget.max_tokens_hot`` — when storing a fragment in the hot - tier would exceed the budget, the least-recently-used fragment(s) are - automatically evicted until the budget is met. If a single fragment - exceeds the entire hot-tier budget it is redirected to the warm tier. - * **Event emission**: Tier transitions (promote, demote, evict) emit - ``DomainEvent`` instances through the optional ``EventBus``. - Event emission is best-effort; a failing bus does not break tier - operations. - - Scope enforcement (``get_scoped``, ``get_scoped_by_resource``, - ``validate_fragment_scope``, ``store_with_scope_check``, - ``get_scoped_metrics``) is provided by ``ScopedTierMixin``. - - Runtime policy (``enforce_staleness``, ``_maybe_auto_promote``, - ``_re_fetch_after_promotion``, ``_enforce_hot_budget``, - ``_emit_tier_event``) is provided by ``TierRuntimeMixin``. + Runtime behaviour (issue #821): auto-promotion on access, staleness + enforcement, budget enforcement on store, and event emission. + Scope enforcement is provided by ``ScopedTierMixin``. + Runtime policy is provided by ``TierRuntimeMixin``. + Settings helpers live in ``context_tier_settings.py``. .. note:: - This service is designed for **single-threaded** use. The - in-memory tier stores are plain ``dict`` instances without - synchronisation. Concurrent callers must coordinate externally. + This service is **thread-safe** (issue #7547). All public methods + acquire ``self._lock`` (a ``threading.RLock``) before accessing the + in-memory tier stores, preventing ``RuntimeError: dictionary changed + size during iteration`` under concurrent plan execution. The + reentrant lock allows internal helpers (e.g. ``promote`` called from + ``_maybe_auto_promote``) to re-acquire the lock without deadlocking. """ def __init__( @@ -99,16 +77,16 @@ class ContextTierService(TierRuntimeMixin, ScopedTierMixin): settings: Settings | None = None, event_bus: EventBus | None = None, ) -> None: - budget = _budget_from_settings(settings) - self._budget = budget + self._budget = budget_from_settings(settings) self._event_bus = event_bus + # --- concurrency lock (issue #7547) -------------------------------- + self._lock = threading.RLock() + # --- runtime policy from settings ---------------------------------- - self._promotion_threshold: int = _promotion_threshold_from_settings( - settings, - ) - self._hot_ttl: timedelta = _hot_ttl_from_settings(settings) - self._warm_ttl: timedelta = _warm_ttl_from_settings(settings) + self._promotion_threshold: int = promotion_threshold_from_settings(settings) + self._hot_ttl = hot_ttl_from_settings(settings) + self._warm_ttl = warm_ttl_from_settings(settings) # --- tier stores --------------------------------------------------- self._hot: dict[str, TieredFragment] = {} @@ -128,18 +106,9 @@ class ContextTierService(TierRuntimeMixin, ScopedTierMixin): def store(self, fragment: TieredFragment) -> None: """Store a fragment in the tier indicated by ``fragment.tier``. - If the fragment already exists in any tier it is moved to the - requested tier. When storing in the hot tier, budget enforcement - is applied: if total hot-tier tokens would exceed - ``TierBudget.max_tokens_hot``, the least-recently-used fragments - are evicted until the budget is satisfied. - - If a single fragment's ``token_count`` exceeds the entire - hot-tier budget, the fragment is redirected to the **warm** tier - instead of being silently evicted after insertion. - - Args: - fragment: The fragment to store. + Moves the fragment if it already exists in another tier. + Enforces hot-tier token budget (LRU eviction). Oversized + fragments (token_count > budget) are redirected to warm. Raises: ValueError: If ``fragment.fragment_id`` is empty. @@ -147,35 +116,36 @@ class ContextTierService(TierRuntimeMixin, ScopedTierMixin): if not fragment.fragment_id: raise ValueError("fragment_id must be non-empty") - # Remove from other tiers first - self._remove_from_all(fragment.fragment_id) + with self._lock: + # Remove from other tiers first + self._remove_from_all(fragment.fragment_id) - if fragment.tier == ContextTier.HOT: - if fragment.token_count > self._budget.max_tokens_hot: - logger.warning( - "tier.oversized_fragment_redirected", - fragment_id=fragment.fragment_id, - token_count=fragment.token_count, - budget=self._budget.max_tokens_hot, - target_tier=ContextTier.WARM.value, - ) - redirected = fragment.model_copy( - update={"tier": ContextTier.WARM}, - ) - self._warm[fragment.fragment_id] = redirected - self._emit_tier_event( - EventType.TIER_DEMOTED, - fragment.fragment_id, - from_tier=ContextTier.HOT, - to_tier=ContextTier.WARM, - ) + if fragment.tier == ContextTier.HOT: + if fragment.token_count > self._budget.max_tokens_hot: + logger.warning( + "tier.oversized_fragment_redirected", + fragment_id=fragment.fragment_id, + token_count=fragment.token_count, + budget=self._budget.max_tokens_hot, + target_tier=ContextTier.WARM.value, + ) + redirected = fragment.model_copy( + update={"tier": ContextTier.WARM}, + ) + self._warm[fragment.fragment_id] = redirected + self._emit_tier_event( + EventType.TIER_DEMOTED, + fragment.fragment_id, + from_tier=ContextTier.HOT, + to_tier=ContextTier.WARM, + ) + else: + self._hot[fragment.fragment_id] = fragment + self._enforce_hot_budget() + elif fragment.tier == ContextTier.WARM: + self._warm[fragment.fragment_id] = fragment else: - self._hot[fragment.fragment_id] = fragment - self._enforce_hot_budget() - elif fragment.tier == ContextTier.WARM: - self._warm[fragment.fragment_id] = fragment - else: - self._cold[fragment.fragment_id] = fragment + self._cold[fragment.fragment_id] = fragment # ------------------------------------------------------------------ # Retrieve @@ -190,28 +160,29 @@ class ContextTierService(TierRuntimeMixin, ScopedTierMixin): Returns ``None`` when the fragment is not found. """ - if fragment_id in self._hot: - self._hot_hit += 1 - frag = self._hot[fragment_id] - return self._touch(frag) + with self._lock: + if fragment_id in self._hot: + self._hot_hit += 1 + frag = self._hot[fragment_id] + return self._touch(frag) - self._hot_miss += 1 + self._hot_miss += 1 - if fragment_id in self._warm: - self._warm_hit += 1 - frag = self._warm[fragment_id] - touched = self._touch(frag) - self._maybe_auto_promote(fragment_id, touched) - return self._re_fetch_after_promotion(fragment_id, touched) + if fragment_id in self._warm: + self._warm_hit += 1 + frag = self._warm[fragment_id] + touched = self._touch(frag) + self._maybe_auto_promote(fragment_id, touched) + return self._re_fetch_after_promotion(fragment_id, touched) - self._warm_miss += 1 + self._warm_miss += 1 - if fragment_id in self._cold: - touched = self._touch(self._cold[fragment_id]) - self._maybe_auto_promote(fragment_id, touched) - return self._re_fetch_after_promotion(fragment_id, touched) + if fragment_id in self._cold: + touched = self._touch(self._cold[fragment_id]) + self._maybe_auto_promote(fragment_id, touched) + return self._re_fetch_after_promotion(fragment_id, touched) - return None + return None def get_all_fragments(self) -> list[TieredFragment]: """Return all fragments across hot/warm/cold tiers. @@ -219,14 +190,16 @@ class ContextTierService(TierRuntimeMixin, ScopedTierMixin): This exposes a stable public read API for callers that need a snapshot across all tiers without reaching into private stores. """ - fragments: list[TieredFragment] = [] - for store in (self._hot, self._warm, self._cold): - fragments.extend(store.values()) - return fragments + with self._lock: + fragments: list[TieredFragment] = [] + for store in (self._hot, self._warm, self._cold): + fragments.extend(store.values()) + return fragments def get_hot_fragments(self) -> list[TieredFragment]: """Return all fragments currently in the hot tier.""" - return list(self._hot.values()) + with self._lock: + return list(self._hot.values()) # ------------------------------------------------------------------ # Actor views @@ -243,30 +216,31 @@ class ContextTierService(TierRuntimeMixin, ScopedTierMixin): ``ScopedBackendView`` for project isolation (when project_names is provided). """ - view = ActorContextView(actor_role=actor_role) - effective_tiers = view.get_effective_tiers() + with self._lock: + view = ActorContextView(actor_role=actor_role) + effective_tiers = view.get_effective_tiers() - candidates: list[TieredFragment] = [] - for tier in effective_tiers: - store = self._store_for_tier(tier) - candidates.extend(store.values()) + candidates: list[TieredFragment] = [] + for tier in effective_tiers: + store = self._store_for_tier(tier) + candidates.extend(store.values()) - # Apply project scoping - if project_names: - scoped = ScopedBackendView( - allowed_projects=frozenset(project_names), - ) - candidates = [f for f in candidates if scoped.is_visible(f)] - logger.debug( - "tier.actor_view_filtered", - actor_role=actor_role.value, - after=len(candidates), - project_names=project_names, - ) + # Apply project scoping + if project_names: + scoped = ScopedBackendView( + allowed_projects=frozenset(project_names), + ) + candidates = [f for f in candidates if scoped.is_visible(f)] + logger.debug( + "tier.actor_view_filtered", + actor_role=actor_role.value, + after=len(candidates), + project_names=project_names, + ) - # Sort by last_accessed descending, cap at max_fragments - candidates.sort(key=lambda f: f.last_accessed, reverse=True) - return candidates[: view.max_fragments] + # Sort by last_accessed descending, cap at max_fragments + candidates.sort(key=lambda f: f.last_accessed, reverse=True) + return candidates[: view.max_fragments] # ------------------------------------------------------------------ # Scoped view @@ -283,19 +257,20 @@ class ContextTierService(TierRuntimeMixin, ScopedTierMixin): """ if not project_names: return [] - scoped = ScopedBackendView( - allowed_projects=frozenset(project_names), - ) - all_frags: list[TieredFragment] = [] - for store in (self._hot, self._warm, self._cold): - all_frags.extend(f for f in store.values() if scoped.is_visible(f)) - all_frags.sort(key=lambda f: f.last_accessed, reverse=True) - logger.debug( - "tier.scoped_view_filtered", - project_names=project_names, - result_count=len(all_frags), - ) - return all_frags + with self._lock: + scoped = ScopedBackendView( + allowed_projects=frozenset(project_names), + ) + all_frags: list[TieredFragment] = [] + for store in (self._hot, self._warm, self._cold): + all_frags.extend(f for f in store.values() if scoped.is_visible(f)) + all_frags.sort(key=lambda f: f.last_accessed, reverse=True) + logger.debug( + "tier.scoped_view_filtered", + project_names=project_names, + result_count=len(all_frags), + ) + return all_frags # ------------------------------------------------------------------ # Promote / Demote @@ -313,98 +288,95 @@ class ContextTierService(TierRuntimeMixin, ScopedTierMixin): if not fragment_id: raise ValueError("fragment_id must be non-empty") - if fragment_id in self._cold: - frag = self._cold.pop(fragment_id) - promoted = frag.model_copy(update={"tier": ContextTier.WARM}) - self._warm[fragment_id] = promoted - self._emit_tier_event( - EventType.TIER_PROMOTED, - fragment_id, - from_tier=ContextTier.COLD, - to_tier=ContextTier.WARM, - ) - return promoted - - if fragment_id in self._warm: - frag = self._warm.pop(fragment_id) - promoted = frag.model_copy(update={"tier": ContextTier.HOT}) - self._hot[fragment_id] = promoted - # Emit PROMOTED before budget enforcement so event ordering - # reflects the causal chain (promotion caused eviction). - self._emit_tier_event( - EventType.TIER_PROMOTED, - fragment_id, - from_tier=ContextTier.WARM, - to_tier=ContextTier.HOT, - ) - self._enforce_hot_budget() - # If budget enforcement evicted the just-promoted fragment, - # restore it to the warm tier to prevent silent data loss. - if fragment_id not in self._hot: - restored = promoted.model_copy( - update={"tier": ContextTier.WARM}, + with self._lock: + if fragment_id in self._cold: + frag = self._cold.pop(fragment_id) + promoted = frag.model_copy(update={"tier": ContextTier.WARM}) + self._warm[fragment_id] = promoted + self._emit_tier_event( + EventType.TIER_PROMOTED, + fragment_id, + from_tier=ContextTier.COLD, + to_tier=ContextTier.WARM, ) - self._warm[fragment_id] = restored - logger.warning( - "tier.promotion_budget_fallback", - fragment_id=fragment_id, - reason="hot budget exceeded after promotion", - ) - return restored - return promoted + return promoted - return None + if fragment_id in self._warm: + frag = self._warm.pop(fragment_id) + promoted = frag.model_copy(update={"tier": ContextTier.HOT}) + self._hot[fragment_id] = promoted + # Emit PROMOTED before budget enforcement so event ordering + # reflects the causal chain (promotion caused eviction). + self._emit_tier_event( + EventType.TIER_PROMOTED, + fragment_id, + from_tier=ContextTier.WARM, + to_tier=ContextTier.HOT, + ) + self._enforce_hot_budget() + # If budget enforcement evicted the just-promoted fragment, + # restore it to the warm tier to prevent silent data loss. + if fragment_id not in self._hot: + restored = promoted.model_copy( + update={"tier": ContextTier.WARM}, + ) + self._warm[fragment_id] = restored + logger.warning( + "tier.promotion_budget_fallback", + fragment_id=fragment_id, + reason="hot budget exceeded after promotion", + ) + return restored + return promoted + + return None def demote(self, fragment_id: str) -> TieredFragment | None: """Demote a fragment one tier down (hot→warm or warm→cold). - When demoting to cold tier, the summarisation hook is applied. + Applies the summarisation hook when demoting to cold. + Resets ``access_count`` to ``0`` so the fragment must + accumulate fresh accesses before re-promotion. Returns the demoted fragment or ``None`` if not found or already in the cold tier. Emits a ``TIER_DEMOTED`` event. - The ``access_count`` is reset to ``0`` on demotion so that - the fragment must accumulate fresh accesses before it can be - auto-promoted again. Without this reset, a previously popular - fragment whose ``access_count`` already exceeds the promotion - threshold would be re-promoted on the very next ``get()`` - call, making staleness enforcement ineffective. - Raises: ValueError: If *fragment_id* is empty. """ if not fragment_id: raise ValueError("fragment_id must be non-empty") - if fragment_id in self._hot: - frag = self._hot.pop(fragment_id) - demoted = frag.model_copy( - update={"tier": ContextTier.WARM, "access_count": 0}, - ) - self._warm[fragment_id] = demoted - self._emit_tier_event( - EventType.TIER_DEMOTED, - fragment_id, - from_tier=ContextTier.HOT, - to_tier=ContextTier.WARM, - ) - return demoted + with self._lock: + if fragment_id in self._hot: + frag = self._hot.pop(fragment_id) + demoted = frag.model_copy( + update={"tier": ContextTier.WARM, "access_count": 0}, + ) + self._warm[fragment_id] = demoted + self._emit_tier_event( + EventType.TIER_DEMOTED, + fragment_id, + from_tier=ContextTier.HOT, + to_tier=ContextTier.WARM, + ) + return demoted - if fragment_id in self._warm: - frag = self._warm.pop(fragment_id) - summarised = self._summarize_for_cold(frag) - demoted = summarised.model_copy( - update={"tier": ContextTier.COLD, "access_count": 0}, - ) - self._cold[fragment_id] = demoted - self._emit_tier_event( - EventType.TIER_DEMOTED, - fragment_id, - from_tier=ContextTier.WARM, - to_tier=ContextTier.COLD, - ) - return demoted + if fragment_id in self._warm: + frag = self._warm.pop(fragment_id) + summarised = self._summarize_for_cold(frag) + demoted = summarised.model_copy( + update={"tier": ContextTier.COLD, "access_count": 0}, + ) + self._cold[fragment_id] = demoted + self._emit_tier_event( + EventType.TIER_DEMOTED, + fragment_id, + from_tier=ContextTier.WARM, + to_tier=ContextTier.COLD, + ) + return demoted - return None + return None # ------------------------------------------------------------------ # LRU eviction @@ -421,25 +393,26 @@ class ContextTierService(TierRuntimeMixin, ScopedTierMixin): if count <= 0: raise ValueError(f"count must be positive, got {count}") - store = self._store_for_tier(tier) - if not store: - return [] + with self._lock: + store = self._store_for_tier(tier) + if not store: + return [] - # Sort by last_accessed ascending (oldest first) - sorted_ids = sorted( - store.keys(), - key=lambda fid: store[fid].last_accessed, - ) - to_evict = sorted_ids[:count] - for fid in to_evict: - del store[fid] - self._emit_tier_event( - EventType.TIER_EVICTED, - fid, - from_tier=tier, - to_tier=None, + # Sort by last_accessed ascending (oldest first) + sorted_ids = sorted( + store.keys(), + key=lambda fid: store[fid].last_accessed, ) - return to_evict + to_evict = sorted_ids[:count] + for fid in to_evict: + del store[fid] + self._emit_tier_event( + EventType.TIER_EVICTED, + fid, + from_tier=tier, + to_tier=None, + ) + return to_evict # ------------------------------------------------------------------ # Metrics @@ -447,15 +420,16 @@ class ContextTierService(TierRuntimeMixin, ScopedTierMixin): def get_metrics(self) -> TierMetrics: """Return current hit/miss and population metrics.""" - return TierMetrics( - hot_count=len(self._hot), - warm_count=len(self._warm), - cold_count=len(self._cold), - hot_hit_count=self._hot_hit, - hot_miss_count=self._hot_miss, - warm_hit_count=self._warm_hit, - warm_miss_count=self._warm_miss, - ) + with self._lock: + return TierMetrics( + hot_count=len(self._hot), + warm_count=len(self._warm), + cold_count=len(self._cold), + hot_hit_count=self._hot_hit, + hot_miss_count=self._hot_miss, + warm_hit_count=self._warm_hit, + warm_miss_count=self._warm_miss, + ) # Scope enforcement methods (get_scoped, get_scoped_by_resource, # validate_fragment_scope, store_with_scope_check, get_scoped_metrics) @@ -475,12 +449,10 @@ class ContextTierService(TierRuntimeMixin, ScopedTierMixin): # ------------------------------------------------------------------ def _find_fragment(self, fragment_id: str) -> TieredFragment | None: - """Look up a fragment across all tiers **without** side effects. + """Look up a fragment across all tiers without side effects. - Unlike :meth:`get`, this does not call :meth:`_touch` and - therefore leaves ``last_accessed``, ``access_count``, and - hit/miss counters unchanged. Used by validation methods that - must be read-only. + Does not call :meth:`_touch`; leaves access metadata unchanged. + Used by read-only validation methods. """ for store in (self._hot, self._warm, self._cold): if fragment_id in store: @@ -524,72 +496,4 @@ class ContextTierService(TierRuntimeMixin, ScopedTierMixin): return fragment.model_copy(update={"content": content}) -# --------------------------------------------------------------------------- -# Budget helper -# --------------------------------------------------------------------------- - - -def _budget_from_settings(settings: Settings | None) -> TierBudget: - """Build a ``TierBudget`` from application settings (or defaults).""" - if settings is None: - return TierBudget() - return TierBudget( - max_tokens_hot=getattr( - settings, "context_max_tokens_hot", _DEFAULT_MAX_TOKENS_HOT - ), - max_decisions_warm=getattr( - settings, "context_max_decisions_warm", _DEFAULT_MAX_DECISIONS_WARM - ), - max_decisions_cold=getattr( - settings, "context_max_decisions_cold", _DEFAULT_MAX_DECISIONS_COLD - ), - ) - - -# --------------------------------------------------------------------------- -# Runtime policy helpers -# --------------------------------------------------------------------------- - - -def _promotion_threshold_from_settings(settings: Settings | None) -> int: - """Read promotion threshold from settings (or use default).""" - if settings is None: - return _DEFAULT_PROMOTION_THRESHOLD - return int( - getattr( - settings, - "context_tier_promotion_threshold", - _DEFAULT_PROMOTION_THRESHOLD, - ), - ) - - -def _hot_ttl_from_settings(settings: Settings | None) -> timedelta: - """Read hot-tier TTL from settings (or use default).""" - if settings is None: - return timedelta(hours=_DEFAULT_HOT_TTL_HOURS) - hours = int( - getattr( - settings, - "context_tier_hot_ttl_hours", - _DEFAULT_HOT_TTL_HOURS, - ), - ) - return timedelta(hours=hours) - - -def _warm_ttl_from_settings(settings: Settings | None) -> timedelta: - """Read warm-tier TTL from settings (or use default).""" - if settings is None: - return timedelta(hours=_DEFAULT_WARM_TTL_HOURS) - hours = int( - getattr( - settings, - "context_tier_warm_ttl_hours", - _DEFAULT_WARM_TTL_HOURS, - ), - ) - return timedelta(hours=hours) - - __all__ = ["ContextTierService"] diff --git a/src/cleveragents/application/services/scoped_tiers.py b/src/cleveragents/application/services/scoped_tiers.py index 8262eb5e5..6ca3a2ff4 100644 --- a/src/cleveragents/application/services/scoped_tiers.py +++ b/src/cleveragents/application/services/scoped_tiers.py @@ -14,6 +14,8 @@ and Forgejo issue #193. from __future__ import annotations +import threading + import structlog from cleveragents.domain.models.acms.scope_resolution import ( @@ -52,6 +54,7 @@ class ScopedTierMixin: _hot_miss: int _warm_hit: int _warm_miss: int + _lock: threading.RLock def get(self, fragment_id: str) -> TieredFragment | None: ... def store(self, fragment: TieredFragment) -> None: ... @@ -103,25 +106,26 @@ class ScopedTierMixin: Args: scope: The resolved resource scope for the plan. """ - scoped = ScopedBackendView.from_resource_scope(scope) - all_frags: list[TieredFragment] = [] - rejected_count = 0 - for store in (self._hot, self._warm, self._cold): - for frag in store.values(): - if scoped.is_visible(frag): - all_frags.append(frag) - else: - rejected_count += 1 - all_frags.sort(key=lambda f: f.last_accessed, reverse=True) - if rejected_count: - logger.info( - "tier.scope_enforcement", - accepted=len(all_frags), - rejected=rejected_count, - project_names=sorted(scope.project_names), - resource_count=len(scope.resource_ids), - ) - return all_frags + with self._lock: + scoped = ScopedBackendView.from_resource_scope(scope) + all_frags: list[TieredFragment] = [] + rejected_count = 0 + for store in (self._hot, self._warm, self._cold): + for frag in store.values(): + if scoped.is_visible(frag): + all_frags.append(frag) + else: + rejected_count += 1 + all_frags.sort(key=lambda f: f.last_accessed, reverse=True) + if rejected_count: + logger.info( + "tier.scope_enforcement", + accepted=len(all_frags), + rejected=rejected_count, + project_names=sorted(scope.project_names), + resource_count=len(scope.resource_ids), + ) + return all_frags def validate_fragment_scope( self, @@ -205,15 +209,16 @@ class ScopedTierMixin: """ if not project_names: raise ValueError("project_names must be non-empty") - scoped = ScopedBackendView( - allowed_projects=frozenset(project_names), - ) - return TierMetrics( - hot_count=sum(1 for f in self._hot.values() if scoped.is_visible(f)), - warm_count=sum(1 for f in self._warm.values() if scoped.is_visible(f)), - cold_count=sum(1 for f in self._cold.values() if scoped.is_visible(f)), - hot_hit_count=self._hot_hit, - hot_miss_count=self._hot_miss, - warm_hit_count=self._warm_hit, - warm_miss_count=self._warm_miss, - ) + with self._lock: + scoped = ScopedBackendView( + allowed_projects=frozenset(project_names), + ) + return TierMetrics( + hot_count=sum(1 for f in self._hot.values() if scoped.is_visible(f)), + warm_count=sum(1 for f in self._warm.values() if scoped.is_visible(f)), + cold_count=sum(1 for f in self._cold.values() if scoped.is_visible(f)), + hot_hit_count=self._hot_hit, + hot_miss_count=self._hot_miss, + warm_hit_count=self._warm_hit, + warm_miss_count=self._warm_miss, + ) diff --git a/src/cleveragents/application/services/tier_runtime.py b/src/cleveragents/application/services/tier_runtime.py index 035d090b3..335d9b581 100644 --- a/src/cleveragents/application/services/tier_runtime.py +++ b/src/cleveragents/application/services/tier_runtime.py @@ -16,6 +16,7 @@ Based on ``docs/specification.md`` ACMS tier sections and issue #821 from __future__ import annotations +import threading from datetime import UTC, datetime, timedelta import structlog @@ -57,6 +58,7 @@ class TierRuntimeMixin: _hot_ttl: timedelta _warm_ttl: timedelta _event_bus: EventBus | None + _lock: threading.RLock def promote(self, fragment_id: str) -> TieredFragment | None: ... def demote(self, fragment_id: str) -> TieredFragment | None: ... @@ -88,46 +90,49 @@ class TierRuntimeMixin: Returns a list of fragment IDs that were demoted. """ - now = datetime.now(tz=UTC) - demoted_ids: list[str] = [] + with self._lock: + now = datetime.now(tz=UTC) + demoted_ids: list[str] = [] - # Snapshot warm tier IDs before hot demotion so we don't - # double-demote fragments in the same pass. - warm_before: set[str] = set(self._warm.keys()) + # Snapshot warm tier IDs before hot demotion so we don't + # double-demote fragments in the same pass. + warm_before: set[str] = set(self._warm.keys()) - # Hot → warm demotion - hot_cutoff = now - self._hot_ttl - stale_hot = [ - fid for fid, frag in self._hot.items() if frag.last_accessed < hot_cutoff - ] - for fid in stale_hot: - self.demote(fid) - demoted_ids.append(fid) - logger.info( - "tier.staleness_demoted", - fragment_id=fid, - from_tier=ContextTier.HOT.value, - to_tier=ContextTier.WARM.value, - ) + # Hot → warm demotion + hot_cutoff = now - self._hot_ttl + stale_hot = [ + fid + for fid, frag in self._hot.items() + if frag.last_accessed < hot_cutoff + ] + for fid in stale_hot: + self.demote(fid) + demoted_ids.append(fid) + logger.info( + "tier.staleness_demoted", + fragment_id=fid, + from_tier=ContextTier.HOT.value, + to_tier=ContextTier.WARM.value, + ) - # Warm → cold demotion (only fragments that were warm BEFORE this pass) - warm_cutoff = now - self._warm_ttl - stale_warm = [ - fid - for fid in warm_before - if fid in self._warm and self._warm[fid].last_accessed < warm_cutoff - ] - for fid in stale_warm: - self.demote(fid) - demoted_ids.append(fid) - logger.info( - "tier.staleness_demoted", - fragment_id=fid, - from_tier=ContextTier.WARM.value, - to_tier=ContextTier.COLD.value, - ) + # Warm → cold demotion (only fragments that were warm BEFORE this pass) + warm_cutoff = now - self._warm_ttl + stale_warm = [ + fid + for fid in warm_before + if fid in self._warm and self._warm[fid].last_accessed < warm_cutoff + ] + for fid in stale_warm: + self.demote(fid) + demoted_ids.append(fid) + logger.info( + "tier.staleness_demoted", + fragment_id=fid, + from_tier=ContextTier.WARM.value, + to_tier=ContextTier.COLD.value, + ) - return demoted_ids + return demoted_ids # ------------------------------------------------------------------ # Auto-promotion (runtime — issue #821) -- 2.52.0 From b43ba41f6dd9d1dceb2fe5e828021dc72cc8cd43 Mon Sep 17 00:00:00 2001 From: HAL9000 Date: Mon, 13 Apr 2026 17:38:35 +0000 Subject: [PATCH 2/2] fix(concurrency): protect validate_fragment_scope with lock and update CONTRIBUTORS - Wrapped validate_fragment_scope() body with self._lock to prevent RuntimeError: dictionary changed size during iteration when another thread mutates the tier stores during scope validation - Updated CONTRIBUTORS.md to document HAL 9000's concurrency safety contributions including thread-safe context tier management (issue #7547) Fixes review feedback from PR #8279. ISSUES CLOSED: #7547 --- CONTRIBUTORS.md | 1 + .../application/services/scoped_tiers.py | 35 ++++++++++--------- 2 files changed, 19 insertions(+), 17 deletions(-) diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index cb09b3a56..056737b45 100644 --- a/CONTRIBUTORS.md +++ b/CONTRIBUTORS.md @@ -16,6 +16,7 @@ Below are some of the specific details of various contributions. * Jeffrey Phillips Freeman has acted as Lead Developer, daily contributor, and Project Owner. * Brent E. Edwards has contributed quality assurance, test coverage, and CI pipeline improvements. * HAL 9000 has contributed automated implementation, bug fixes, and feature development as part of the CleverAgents automation pool. +* HAL 9000 has contributed concurrency safety improvements, including thread-safe context tier management (issue #7547) for parallel plan execution. * HAL 9000 has contributed the plan concurrency race-condition fix (#7989): wired `LockService` into the plan lifecycle, guarding `execute_plan()` and `apply_plan()` with plan-level advisory locks and unique per-invocation owner identities to prevent silent concurrent state corruption. * This project was made possible thanks to considerable donation of time, money, and resources by CleverThis, Inc. * HAL 9000 has contributed automated bug fixes, CLI output formatting improvements, and ongoing maintenance as part of the CleverAgents automation system. diff --git a/src/cleveragents/application/services/scoped_tiers.py b/src/cleveragents/application/services/scoped_tiers.py index 6ca3a2ff4..2edc5155b 100644 --- a/src/cleveragents/application/services/scoped_tiers.py +++ b/src/cleveragents/application/services/scoped_tiers.py @@ -138,24 +138,25 @@ class ScopedTierMixin: ScopeViolationError: If the fragment is out of scope. ValueError: If the fragment does not exist. """ - frag = self._find_fragment(fragment_id) - if frag is None: - raise ValueError(f"Fragment not found: {fragment_id}") + with self._lock: + frag = self._find_fragment(fragment_id) + if frag is None: + raise ValueError(f"Fragment not found: {fragment_id}") - scoped = ScopedBackendView.from_resource_scope(scope) - if not scoped.is_visible(frag): - logger.warning( - "tier.fragment_scope_violation", - fragment_id=fragment_id, - fragment_project=frag.project_name, - fragment_resource=frag.resource_id, - scope_projects=sorted(scope.project_names), - ) - raise ScopeViolationError( - f"Fragment '{fragment_id}' is not within the active scope", - resource_ids=(frag.resource_id,) if frag.resource_id else (), - scope_project_names=tuple(sorted(scope.project_names)), - ) + scoped = ScopedBackendView.from_resource_scope(scope) + if not scoped.is_visible(frag): + logger.warning( + "tier.fragment_scope_violation", + fragment_id=fragment_id, + fragment_project=frag.project_name, + fragment_resource=frag.resource_id, + scope_projects=sorted(scope.project_names), + ) + raise ScopeViolationError( + f"Fragment '{fragment_id}' is not within the active scope", + resource_ids=(frag.resource_id,) if frag.resource_id else (), + scope_project_names=tuple(sorted(scope.project_names)), + ) def store_with_scope_check( self, -- 2.52.0