"""Step definitions for context tier runtime promotion/demotion/eviction. Covers the runtime logic added by issue #821: auto-promotion on access, time-based staleness enforcement, budget-based eviction on store, and tier transition event emission via the ``EventBus``. """ from __future__ import annotations from collections.abc import Callable 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, TierBudget, TieredFragment, ) from cleveragents.infrastructure.events.models import DomainEvent from cleveragents.infrastructure.events.types import EventType __all__: list[str] = [] # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_fragment( fragment_id: str, tier: ContextTier, token_count: int = 50, content: str = "test content", last_accessed: datetime | None = None, ) -> TieredFragment: """Create a ``TieredFragment`` with the given properties.""" frag = TieredFragment( fragment_id=fragment_id, content=content, tier=tier, token_count=token_count, project_name="test-project", ) if last_accessed is not None: frag.last_accessed = last_accessed return frag class _EventCollector: """Simple event bus replacement that collects emitted events.""" def __init__(self) -> None: self.events: list[DomainEvent] = [] def emit(self, event: DomainEvent) -> None: self.events.append(event) def subscribe( self, event_type: EventType, handler: Callable[[DomainEvent], None], ) -> None: pass # Not needed for test assertions # --------------------------------------------------------------------------- # Background / Given steps # --------------------------------------------------------------------------- @given("a context tier service with default settings") def step_service_defaults(context: Any) -> None: context.tier_service = ContextTierService(settings=None) @given('a fragment "{fid}" stored in the cold tier with {tokens:d} tokens') def step_store_cold(context: Any, fid: str, tokens: int) -> None: frag = _make_fragment(fid, ContextTier.COLD, token_count=tokens) context.tier_service.store(frag) @given('a fragment "{fid}" stored in the warm tier with {tokens:d} tokens') def step_store_warm(context: Any, fid: str, tokens: int) -> None: frag = _make_fragment(fid, ContextTier.WARM, token_count=tokens) context.tier_service.store(frag) @given('a fragment "{fid}" stored in the hot tier with {tokens:d} tokens') def step_store_hot(context: Any, fid: str, tokens: int) -> None: frag = _make_fragment(fid, ContextTier.HOT, token_count=tokens) context.tier_service.store(frag) @given( 'a fragment "{fid}" stored in the hot tier with stale timestamp of {hours:d} hours' ) def step_store_hot_stale(context: Any, fid: str, hours: int) -> None: stale = datetime.now(tz=UTC) - timedelta(hours=hours) frag = _make_fragment(fid, ContextTier.HOT, last_accessed=stale) context.tier_service.store(frag) @given('a fragment "{fid}" stored in the hot tier with fresh timestamp') def step_store_hot_fresh(context: Any, fid: str) -> None: frag = _make_fragment(fid, ContextTier.HOT) context.tier_service.store(frag) @given( 'a fragment "{fid}" stored in the warm tier with stale timestamp of {hours:d} hours' ) def step_store_warm_stale(context: Any, fid: str, hours: int) -> None: stale = datetime.now(tz=UTC) - timedelta(hours=hours) frag = _make_fragment(fid, ContextTier.WARM, last_accessed=stale) context.tier_service.store(frag) @given("a service with hot tier budget of {tokens:d} tokens") def step_service_custom_budget(context: Any, tokens: int) -> None: svc = ContextTierService(settings=None) svc._budget = TierBudget(max_tokens_hot=tokens) context.tier_service = svc @given( 'a fragment "{fid}" stored in the hot tier with {tokens:d} tokens and old timestamp' ) def step_store_hot_old(context: Any, fid: str, tokens: int) -> None: old = datetime.now(tz=UTC) - timedelta(minutes=10) frag = _make_fragment(fid, ContextTier.HOT, token_count=tokens, last_accessed=old) context.tier_service.store(frag) @given( 'a fragment "{fid}" stored in the hot tier' " with {tokens:d} tokens and recent timestamp" ) def step_store_hot_recent(context: Any, fid: str, tokens: int) -> None: frag = _make_fragment(fid, ContextTier.HOT, token_count=tokens) context.tier_service.store(frag) @given("a context tier service with an event bus") def step_service_with_bus(context: Any) -> None: collector = _EventCollector() svc = ContextTierService(settings=None, event_bus=collector) context.tier_service = svc context.event_collector = collector @given("a context tier service with an event bus and hot budget of {tokens:d} tokens") def step_service_with_bus_budget(context: Any, tokens: int) -> None: collector = _EventCollector() svc = ContextTierService(settings=None, event_bus=collector) svc._budget = TierBudget(max_tokens_hot=tokens) context.tier_service = svc context.event_collector = collector # --------------------------------------------------------------------------- # When steps # --------------------------------------------------------------------------- @when('I access "{fid}" via get {n:d} times') def step_access_n(context: Any, fid: str, n: int) -> None: svc: ContextTierService = context.tier_service for _ in range(n): svc.get(fid) @when("I invoke enforce_staleness on the service") def step_enforce_staleness(context: Any) -> None: svc: ContextTierService = context.tier_service context.staleness_result = svc.enforce_staleness() @when('I store fragment "{fid}" in the hot tier with {tokens:d} tokens') def step_store_hot_overflow(context: Any, fid: str, tokens: int) -> None: frag = _make_fragment(fid, ContextTier.HOT, token_count=tokens) context.tier_service.store(frag) @when('I promote fragment "{fid}"') def step_promote(context: Any, fid: str) -> None: context.tier_service.promote(fid) @when('I demote fragment "{fid}"') def step_demote(context: Any, fid: str) -> None: context.tier_service.demote(fid) @when("I evict {count:d} LRU fragment from the hot tier") def step_evict_lru_hot(context: Any, count: int) -> None: context.tier_service.evict_lru(ContextTier.HOT, count) # --------------------------------------------------------------------------- # Then steps # --------------------------------------------------------------------------- @then('fragment "{fid}" should be in the warm tier') def step_in_warm(context: Any, fid: str) -> None: svc: ContextTierService = context.tier_service frag = svc._find_fragment(fid) assert frag is not None, f"Fragment '{fid}' not found in any tier" assert frag.tier == ContextTier.WARM, f"Expected '{fid}' in WARM, got {frag.tier}" @then('fragment "{fid}" should be in the hot tier') def step_in_hot(context: Any, fid: str) -> None: svc: ContextTierService = context.tier_service frag = svc._find_fragment(fid) assert frag is not None, f"Fragment '{fid}' not found in any tier" assert frag.tier == ContextTier.HOT, f"Expected '{fid}' in HOT, got {frag.tier}" @then('fragment "{fid}" should be in the cold tier') def step_in_cold(context: Any, fid: str) -> None: svc: ContextTierService = context.tier_service frag = svc._find_fragment(fid) assert frag is not None, f"Fragment '{fid}' not found in any tier" assert frag.tier == ContextTier.COLD, f"Expected '{fid}' in COLD, got {frag.tier}" @then('fragment "{fid}" should not be in any tier') def step_not_in_any(context: Any, fid: str) -> None: svc: ContextTierService = context.tier_service frag = svc._find_fragment(fid) assert frag is None, f"Expected '{fid}' to be evicted, but found in {frag.tier}" @then("the staleness result should contain {n:d} fragment IDs") def step_staleness_count(context: Any, n: int) -> None: result: list[str] = context.staleness_result assert len(result) == n, ( f"Expected {n} demoted fragments, got {len(result)}: {result}" ) @then('a TIER_PROMOTED event should have been emitted for "{fid}"') def step_promoted_event(context: Any, fid: str) -> None: collector: _EventCollector = context.event_collector promoted = [ e for e in collector.events if e.event_type == EventType.TIER_PROMOTED and e.details.get("fragment_id") == fid ] assert len(promoted) >= 1, ( f"No TIER_PROMOTED event for '{fid}'. " f"Events: {[e.event_type for e in collector.events]}" ) @then('a TIER_DEMOTED event should have been emitted for "{fid}"') def step_demoted_event(context: Any, fid: str) -> None: collector: _EventCollector = context.event_collector demoted = [ e for e in collector.events if e.event_type == EventType.TIER_DEMOTED and e.details.get("fragment_id") == fid ] assert len(demoted) >= 1, ( f"No TIER_DEMOTED event for '{fid}'. " f"Events: {[e.event_type for e in collector.events]}" ) @then('a TIER_EVICTED event should have been emitted for "{fid}"') def step_evicted_event(context: Any, fid: str) -> None: collector: _EventCollector = context.event_collector evicted = [ e for e in collector.events if e.event_type == EventType.TIER_EVICTED and e.details.get("fragment_id") == fid ] assert len(evicted) >= 1, ( f"No TIER_EVICTED event for '{fid}'. " f"Events: {[e.event_type for e in collector.events]}" ) @then( 'the last TIER_PROMOTED event should have from_tier "{from_t}" and to_tier "{to_t}"' ) def step_promoted_event_details(context: Any, from_t: str, to_t: str) -> None: collector: _EventCollector = context.event_collector promoted = [e for e in collector.events if e.event_type == EventType.TIER_PROMOTED] assert len(promoted) >= 1, "No TIER_PROMOTED events" last = promoted[-1] assert last.details["from_tier"] == from_t, ( f"Expected from_tier='{from_t}', got '{last.details['from_tier']}'" ) assert last.details["to_tier"] == to_t, ( f"Expected to_tier='{to_t}', got '{last.details['to_tier']}'" ) @then( 'the last TIER_DEMOTED event should have from_tier "{from_t}" and to_tier "{to_t}"' ) def step_demoted_event_details(context: Any, from_t: str, to_t: str) -> None: collector: _EventCollector = context.event_collector demoted = [e for e in collector.events if e.event_type == EventType.TIER_DEMOTED] assert len(demoted) >= 1, "No TIER_DEMOTED events" last = demoted[-1] assert last.details["from_tier"] == from_t, ( f"Expected from_tier='{from_t}', got '{last.details['from_tier']}'" ) assert last.details["to_tier"] == to_t, ( f"Expected to_tier='{to_t}', got '{last.details['to_tier']}'" ) class _FailingBus: """Event bus stub that raises on every emit() call.""" def emit(self, event: DomainEvent) -> None: msg = "Simulated event bus failure" raise RuntimeError(msg) def subscribe( self, event_type: EventType, handler: Callable[[DomainEvent], None], ) -> None: pass @given("a context tier service with a failing event bus") def step_service_with_failing_bus(context: Any) -> None: bus = _FailingBus() svc = ContextTierService(settings=None, event_bus=bus) context.tier_service = svc @given( 'a fragment "{fid}" stored in the warm tier' " with {tokens:d} tokens and old timestamp" ) def step_store_warm_old(context: Any, fid: str, tokens: int) -> None: old = datetime.now(tz=UTC) - timedelta(minutes=10) frag = _make_fragment(fid, ContextTier.WARM, token_count=tokens, last_accessed=old) context.tier_service.store(frag) @when("I invoke enforce_staleness on the service with a hot TTL of {hours:d} hours") def step_enforce_staleness_custom_ttl(context: Any, hours: int) -> None: svc: ContextTierService = context.tier_service svc._hot_ttl = timedelta(hours=hours) context.staleness_result = svc.enforce_staleness() @given("a context tier service with promotion threshold {threshold:d}") def step_service_custom_threshold(context: Any, threshold: int) -> None: svc = ContextTierService(settings=None) svc._promotion_threshold = threshold context.tier_service = svc