"""Step definitions for TDD Issue #821 — context tier runtime logic. These steps exercise ``ContextTierService`` and verify that it automatically promotes fragments on repeated access, demotes stale fragments, and evicts fragments when the hot tier budget overflows. Bug #821 identified that ``ContextTierService`` had data models and manual ``promote()``/``demote()``/``evict_lru()`` methods but NO automatic runtime logic. The bug has been fixed — these tests now serve as a permanent regression guard confirming the runtime logic works correctly. """ from __future__ import annotations from datetime import UTC, datetime, timedelta from behave import given, then, when from behave.runner import Context from cleveragents.application.services.context_tiers import ContextTierService from cleveragents.domain.models.acms.tiers import ( ContextTier, TierBudget, TieredFragment, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _FRAGMENT_ID: str = "frag-tier-runtime-001" _STALE_FRAGMENT_ID: str = "frag-stale-hot-001" 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.""" kwargs: dict[str, object] = { "fragment_id": fragment_id, "content": content, "tier": tier, "token_count": token_count, "project_name": "test-project", } if last_accessed is not None: kwargs["last_accessed"] = last_accessed return TieredFragment(**kwargs) # type: ignore[arg-type] # --------------------------------------------------------------------------- # Scenario 1: Promotion on repeated access # --------------------------------------------------------------------------- @given("a context tier service with default budget") def step_service_default_budget(context: Context) -> None: """Create a ContextTierService with the default TierBudget.""" context.tier_service = ContextTierService(settings=None) @given("a fragment stored in the cold tier") def step_fragment_in_cold(context: Context) -> None: """Store a fragment in the cold tier.""" frag: TieredFragment = _make_fragment( fragment_id=_FRAGMENT_ID, tier=ContextTier.COLD, token_count=50, ) service: ContextTierService = context.tier_service service.store(frag) # Verify it is actually in cold tier stored: TieredFragment | None = service.get(_FRAGMENT_ID) assert stored is not None, "Fragment was not stored" assert stored.tier == ContextTier.COLD, ( f"Fragment should be in cold tier, got {stored.tier}" ) context.fragment_id = _FRAGMENT_ID @when("I access the fragment {n:d} times via get") def step_access_n_times(context: Context, n: int) -> None: """Access the fragment N times through the get() method.""" service: ContextTierService = context.tier_service frag_id: str = context.fragment_id for _ in range(n): result: TieredFragment | None = service.get(frag_id) assert result is not None, "Fragment disappeared during access" context.accessed_fragment = service.get(frag_id) @then("the fragment should have been promoted to warm or hot tier") def step_fragment_promoted(context: Context) -> None: """Assert the fragment was automatically promoted after repeated access. Bug #821: ``get()`` increments ``access_count`` and updates ``last_accessed`` but never calls ``promote()`` or any equivalent runtime tier-transition logic. The fragment remains in the cold tier regardless of how many times it is accessed. """ service: ContextTierService = context.tier_service frag: TieredFragment | None = service.get(context.fragment_id) assert frag is not None, "Fragment not found" # The fragment should have been auto-promoted out of cold tier assert frag.tier != ContextTier.COLD, ( f"Fragment is still in COLD tier after repeated access " f"(access_count={frag.access_count}). ContextTierService.get() " f"updates access metadata but does not auto-promote fragments " f"based on access patterns (bug #821). Expected tier to be " f"WARM or HOT." ) # --------------------------------------------------------------------------- # Scenario 2: Demotion on staleness # --------------------------------------------------------------------------- @given("a fragment stored in the hot tier with a stale last_accessed timestamp") def step_stale_hot_fragment(context: Context) -> None: """Store a fragment in the hot tier with a very old last_accessed.""" stale_time: datetime = datetime.now(tz=UTC) - timedelta(hours=24) frag: TieredFragment = _make_fragment( fragment_id=_STALE_FRAGMENT_ID, tier=ContextTier.HOT, token_count=100, last_accessed=stale_time, ) service: ContextTierService = context.tier_service service.store(frag) context.stale_fragment_id = _STALE_FRAGMENT_ID @when("I invoke the staleness enforcement runtime") def step_invoke_staleness(context: Context) -> None: """Attempt to invoke the staleness enforcement method. Bug #821: There is no method on ContextTierService that enforces staleness-based demotion. Methods like ``enforce_staleness()``, ``apply_tier_policy()``, or ``tick()`` do not exist. We attempt to call the most likely candidate names; if none exist, we record the failure. """ service: ContextTierService = context.tier_service context.staleness_invoked = False # Try plausible method names for staleness enforcement for method_name in ( "enforce_staleness", "apply_tier_policy", "tick", "enforce_demotion_policy", "run_lifecycle", "apply_lifecycle", ): method = getattr(service, method_name, None) if callable(method): method() context.staleness_invoked = True return # No runtime method found — record this for the assertion step context.staleness_invoked = False @then("the fragment should have been demoted to warm or cold tier") def step_fragment_demoted(context: Context) -> None: """Assert the stale fragment was automatically demoted. Bug #821: ContextTierService has no staleness enforcement runtime. There is no method that inspects ``last_accessed`` timestamps and auto-demotes stale hot-tier fragments. The fragment stays in the hot tier indefinitely. """ service: ContextTierService = context.tier_service # First check: was there even a method to invoke? assert context.staleness_invoked, ( "No staleness enforcement method found on ContextTierService. " "Tried: enforce_staleness(), apply_tier_policy(), tick(), " "enforce_demotion_policy(), run_lifecycle(), apply_lifecycle(). " "ContextTierService has no runtime logic for automatic demotion " "of stale fragments (bug #821)." ) # Second check: did the fragment actually move? frag: TieredFragment | None = service.get(context.stale_fragment_id) assert frag is not None, "Stale fragment not found" assert frag.tier != ContextTier.HOT, ( f"Stale fragment is still in HOT tier " f"(last_accessed={frag.last_accessed}). " f"ContextTierService has no runtime staleness enforcement — " f"stale fragments are never automatically demoted (bug #821). " f"Expected tier to be WARM or COLD." ) # --------------------------------------------------------------------------- # Scenario 3: Eviction on budget overflow # --------------------------------------------------------------------------- @given("a context tier service with a small hot tier budget of {n:d} tokens") def step_service_small_budget(context: Context, n: int) -> None: """Create a ContextTierService with a constrained hot tier budget.""" # ContextTierService accepts Settings; we pass None and override budget service: ContextTierService = ContextTierService(settings=None) service._budget = TierBudget( max_tokens_hot=n, max_decisions_warm=500, max_decisions_cold=5000, ) context.tier_service = service context.hot_budget_tokens = n @given("the hot tier is filled to its token budget limit") def step_fill_hot_tier(context: Context) -> None: """Fill the hot tier to exactly the token budget limit.""" service: ContextTierService = context.tier_service budget_tokens: int = context.hot_budget_tokens # Store fragments totalling exactly the budget limit # Use 50-token fragments so 100 tokens = 2 fragments frag_size: int = 50 num_frags: int = budget_tokens // frag_size context.initial_hot_fragment_ids = [] for i in range(num_frags): frag_id: str = f"frag-budget-{i:03d}" frag: TieredFragment = _make_fragment( fragment_id=frag_id, tier=ContextTier.HOT, token_count=frag_size, content="x" * frag_size, last_accessed=datetime.now(tz=UTC) - timedelta(minutes=num_frags - i), ) service.store(frag) context.initial_hot_fragment_ids.append(frag_id) # Record the oldest fragment (lowest last_accessed) context.oldest_fragment_id = context.initial_hot_fragment_ids[0] metrics = service.get_metrics() assert metrics.hot_count == num_frags, ( f"Expected {num_frags} fragments in hot tier, got {metrics.hot_count}" ) @when("I store one more fragment in the hot tier") def step_store_overflow(context: Context) -> None: """Store one more fragment that would exceed the hot tier budget.""" service: ContextTierService = context.tier_service overflow_frag: TieredFragment = _make_fragment( fragment_id="frag-overflow-001", tier=ContextTier.HOT, token_count=50, content="overflow content", ) service.store(overflow_frag) context.overflow_fragment_id = "frag-overflow-001" @then("the oldest fragment should have been evicted from the hot tier") def step_oldest_evicted(context: Context) -> None: """Assert the oldest hot-tier fragment was automatically evicted. Bug #821: ``store()`` does not check ``TierBudget.max_tokens_hot`` and does not trigger ``evict_lru()`` when the budget is exceeded. The hot tier grows without bound. """ service: ContextTierService = context.tier_service budget_tokens: int = context.hot_budget_tokens # Calculate total tokens now in hot tier hot_total_tokens: int = sum(f.token_count for f in service._hot.values()) # The hot tier should not exceed its budget assert hot_total_tokens <= budget_tokens, ( f"Hot tier has {hot_total_tokens} tokens but budget is " f"{budget_tokens} tokens. store() does not enforce " f"TierBudget.max_tokens_hot or trigger automatic LRU " f"eviction when the budget is exceeded (bug #821)." ) # The oldest fragment should have been evicted oldest_id: str = context.oldest_fragment_id oldest_frag: TieredFragment | None = service.get(oldest_id) assert oldest_frag is None or oldest_frag.tier != ContextTier.HOT, ( f"Oldest fragment '{oldest_id}' is still in the hot tier after " f"budget overflow. ContextTierService.store() does not auto-evict " f"the least-recently-used fragment when hot tier budget is " f"exceeded (bug #821)." )