cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

fix(concurrency): add thread safety to ContextTierService

Browse Source 
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
This commit is contained in:
HAL 9000
2026-04-13 07:29:00 +00:00
parent b752dd485f
commit 0f75429253
7 changed files with 967 additions and 390 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+15
View File
@@ -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
features/context_tier_thread_safety.feature
+84
View File
@@ -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
features/steps/context_tier_thread_safety_steps.py
+461
View File
@@ -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}"
src/cleveragents/application/services/context_tier_settings.py
+103
View File
@@ -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",
]
src/cleveragents/application/services/context_tiers.py
+227 -323
View File
@@ -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"]
src/cleveragents/application/services/scoped_tiers.py
+36 -31
View File
@@ -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,
)
src/cleveragents/application/services/tier_runtime.py
+41 -36
View File
@@ -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)