From 34c2acc354d6a98863580f3a298a25bbcf4640be Mon Sep 17 00:00:00 2001 From: Luis Mendes Date: Tue, 24 Mar 2026 15:06:57 +0000 Subject: [PATCH] fix(acms): implement context tier runtime promotion/demotion/eviction Implement the missing runtime logic for the ACMS context tier service. Previously only data models and manual promote()/demote()/evict_lru() methods existed. This commit adds: - Auto-promotion on access: get() now promotes fragments one tier up when access_count reaches the configurable promotion_threshold (default: 5 accesses). The access counter resets after each successful promotion so fragments must accumulate fresh accesses before the next tier transition. - Staleness enforcement: new enforce_staleness() method demotes hot fragments older than hot_ttl (default: 24h) to warm, and warm fragments older than warm_ttl to cold. A snapshot of existing warm-tier IDs prevents double-demotion in a single pass. - Budget enforcement on store and promote: store() and promote() now enforce TierBudget.max_tokens_hot by evicting LRU hot-tier fragments until the token budget is met. The eviction loop uses incremental token tracking to avoid recomputing the sum. - Event emission: Added TIER_PROMOTED, TIER_DEMOTED, TIER_EVICTED event types to EventType enum. All tier transitions emit DomainEvent instances through the optional EventBus. - Configuration: Added context_tier_promotion_threshold, context_tier_hot_ttl_hours, context_tier_warm_ttl_hours settings. The warm TTL setting also accepts the spec-defined env var CLEVERAGENTS_CTX_WARM_HOURS as an alias. - DI wiring: container.py now injects event_bus into context_tier_service. Review fixes applied (code review on PR #1150): - C1: Reset access_count to 0 after each auto-promotion to prevent chain promotion that bypassed the warm tier. - C2: Call _enforce_hot_budget() inside promote() warm-to-hot path so auto-promoted fragments respect the token budget. - H1: Corrected _enforce_hot_budget() docstring: actual complexity is O(n + n*k) not O(n), since min() scans remaining entries on each eviction. - M1: Added CLEVERAGENTS_CTX_WARM_HOURS as an additional env var alias for context_tier_warm_ttl_hours per specification line 30555. Review fixes applied (second code review on PR #1150): - B-CRIT-1: Fixed data loss in promote() warm-to-hot: emit TIER_PROMOTED before _enforce_hot_budget(), and if the promoted fragment is evicted by budget, restore it to the warm tier instead of silently losing it. - B-HIGH-1: Fixed self-eviction on store(): fragments whose token_count exceeds the entire hot-tier budget are now redirected to the warm tier with a warning log. - B-MED-1: Wrapped _emit_tier_event() in try/except so a failing event bus does not break tier operations (best-effort emission). - B-MED-2: Fixed event ordering so TIER_PROMOTED fires before any budget-triggered TIER_EVICTED events. - D-LOW-1: Fixed type hint in Robot helper (dict[str, Callable]). - D-LOW-2: Added __all__ export to context_tiers.py. - S-LOW-1: Added thread-safety docstring note to ContextTierService. Review fixes applied (third code review on PR #1150): - B-MED-1: Added TIER_DEMOTED event emission for oversized fragment redirect in store(), closing the observability gap where the only tier transition without event emission was the hot-to-warm redirect for fragments exceeding the entire hot-tier budget. - S-LOW-1: Added CLEVERAGENTS_CTX_HOT_HOURS as an additional env var alias for context_tier_hot_ttl_hours, for consistency with the warm-tier alias CLEVERAGENTS_CTX_WARM_HOURS. - S-LOW-2: Added docstring note to enforce_staleness() reconciling the hot-tier TTL with the specification statement that hot-tier retention is "Until resource removed" (TTL controls tier placement, not data retention). Review fixes applied (fourth code review on PR #1150): - B-HIGH-1: Reset access_count to 0 on demotion so that demoted fragments must accumulate fresh accesses before re-promotion. Without this reset, a previously popular fragment whose access_count already exceeded the promotion threshold would be re-promoted on the very next get() call, making staleness enforcement ineffective. Review fixes applied (freemo APPROVED review on PR #1150): - #1: Removed all # type: ignore annotations from test files. Fixed _EventCollector, _FailingBus, and _NullBus subscribe() signatures to use Callable[[DomainEvent], None] matching the EventBus protocol. Replaced dict-spread TieredFragment construction with explicit keyword arguments and post-construction assignment. - #2: Extracted runtime policy logic (enforce_staleness, _maybe_auto_promote, _re_fetch_after_promotion, _enforce_hot_budget, _emit_tier_event) into TierRuntimeMixin in tier_runtime.py to reduce context_tiers.py toward the 500-line guideline. - #3: Added fragment_id non-empty validation guard to promote() and demote() per CONTRIBUTING.md argument validation policy. - #8: Renamed _resolve to _re_fetch_after_promotion for clarity. Removed @tdd_expected_fail from TDD tests (Behave + Robot) as the bug is now fixed. All 3 TDD scenarios pass normally. Tests: 27 Behave scenarios (24 feature + 3 TDD), 4 Robot integration tests, 4 ASV benchmark suites. ISSUES CLOSED: #821 --- CHANGELOG.md | 18 + benchmarks/bench_context_tier_runtime.py | 139 +++++++ features/context_tier_runtime.feature | 179 +++++++++ features/steps/context_tier_runtime_steps.py | 367 ++++++++++++++++++ features/tdd_context_tier_runtime.feature | 2 +- robot/context_tier_runtime.robot | 43 ++ robot/helper_context_tier_runtime.py | 158 ++++++++ robot/tdd_context_tier_runtime.robot | 6 +- src/cleveragents/application/container.py | 2 + .../application/services/context_tiers.py | 233 ++++++++++- .../application/services/tier_runtime.py | 251 ++++++++++++ src/cleveragents/config/settings.py | 26 ++ .../infrastructure/events/types.py | 5 + 13 files changed, 1413 insertions(+), 16 deletions(-) create mode 100644 benchmarks/bench_context_tier_runtime.py create mode 100644 features/context_tier_runtime.feature create mode 100644 features/steps/context_tier_runtime_steps.py create mode 100644 robot/context_tier_runtime.robot create mode 100644 robot/helper_context_tier_runtime.py create mode 100644 src/cleveragents/application/services/tier_runtime.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ff5d522fe..0fdf7bcce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,24 @@ pathlib/os/difflib. DevcontainerHandler implements read, write, and discover_children via `devcontainer exec`. DatabaseResourceHandler inherits NotImplementedError stubs pending connection management. (#827) +- Implemented ACMS context tier runtime promotion/demotion/eviction: + auto-promotion on access with configurable threshold (default: 5), + time-based staleness enforcement (hot/warm TTL, default: 24h each), + budget-based LRU eviction on hot-tier overflow, and tier transition + event emission (TIER_PROMOTED, TIER_DEMOTED, TIER_EVICTED) via + EventBus. Added `context_tier_promotion_threshold`, + `context_tier_hot_ttl_hours`, and `context_tier_warm_ttl_hours` + settings with DI wiring of event_bus into ContextTierService. + Oversized fragments that exceed the entire hot-tier budget are now + redirected to the warm tier with a TIER_DEMOTED event. Promotion + to hot falls back to warm when the promoted fragment is evicted by + budget enforcement. Event emission is best-effort; a failing event + bus no longer breaks tier operations. Added `CLEVERAGENTS_CTX_HOT_HOURS` + env var alias for `context_tier_hot_ttl_hours` for consistency with + the warm-tier alias. Demotion now resets `access_count` to zero so + that demoted fragments must accumulate fresh accesses before + re-promotion, preventing staleness enforcement from being immediately + undone by a single access. (#821) - Aligned plan lifecycle model with specification: ERRORED is now terminal in `is_terminal`, per-phase state validation enforces APPLIED/CONSTRAINED to APPLY-only and COMPLETE to diff --git a/benchmarks/bench_context_tier_runtime.py b/benchmarks/bench_context_tier_runtime.py new file mode 100644 index 000000000..67e1cc040 --- /dev/null +++ b/benchmarks/bench_context_tier_runtime.py @@ -0,0 +1,139 @@ +"""ASV benchmarks for context tier runtime logic (issue #821). + +Measures the overhead of auto-promotion on access, staleness +enforcement, and budget-based eviction. +""" + +from __future__ import annotations + +import importlib +import sys +from datetime import UTC, datetime, timedelta +from pathlib import Path + +# Ensure source tree is importable +_SRC = str(Path(__file__).resolve().parents[1] / "src") +if _SRC not in sys.path: + sys.path.insert(0, _SRC) +import cleveragents # noqa: E402 + +importlib.reload(cleveragents) + +from cleveragents.application.services.context_tiers import ( # noqa: E402 + ContextTierService, +) +from cleveragents.domain.models.acms.tiers import ( # noqa: E402 + ContextTier, + TierBudget, + TieredFragment, +) + + +def _make_frag( + fid: str, + tier: ContextTier, + tokens: int = 50, + last_accessed: datetime | None = None, +) -> TieredFragment: + frag = TieredFragment( + fragment_id=fid, + content="bench", + tier=tier, + token_count=tokens, + project_name="bench", + ) + if last_accessed is not None: + frag.last_accessed = last_accessed + return frag + + +class AutoPromotionSuite: + """Benchmark auto-promotion on repeated ``get()`` calls.""" + + timeout = 60 + + def setup(self) -> None: + self.service = ContextTierService(settings=None) + frag = _make_frag("bench-promote", ContextTier.COLD) + self.service.store(frag) + + def time_auto_promote_5_accesses(self) -> None: + for _ in range(5): + self.service.get("bench-promote") + + +class StalenessEnforcementSuite: + """Benchmark ``enforce_staleness()`` with varying fragment counts.""" + + timeout = 60 + + def setup(self) -> None: + self.service = ContextTierService(settings=None) + stale = datetime.now(tz=UTC) - timedelta(hours=25) + for i in range(100): + frag = _make_frag(f"stale-{i:04d}", ContextTier.HOT, last_accessed=stale) + self.service.store(frag) + + def time_enforce_staleness_100_fragments(self) -> None: + self.service.enforce_staleness() + + +class BudgetEvictionSuite: + """Benchmark budget-based eviction on hot-tier overflow.""" + + timeout = 60 + + def setup(self) -> None: + self.service = ContextTierService(settings=None) + self.service._budget = TierBudget(max_tokens_hot=500) + old_time = datetime.now(tz=UTC) - timedelta(minutes=100) + for i in range(10): + ts = old_time + timedelta(minutes=i) + frag = _make_frag( + f"budget-{i:03d}", + ContextTier.HOT, + tokens=50, + last_accessed=ts, + ) + self.service.store(frag) + self.overflow_frag = _make_frag( + "overflow", + ContextTier.HOT, + tokens=50, + ) + + def time_store_with_budget_eviction(self) -> None: + self.service.store(self.overflow_frag) + + +class TierEventEmissionSuite: + """Benchmark tier transition event emission overhead.""" + + timeout = 60 + + def setup(self) -> None: + from collections.abc import Callable + + from cleveragents.infrastructure.events.models import DomainEvent + from cleveragents.infrastructure.events.types import EventType + + class _NullBus: + def emit(self, event: DomainEvent) -> None: + pass + + def subscribe( + self, + event_type: EventType, + handler: Callable[[DomainEvent], None], + ) -> None: + pass + + self.service = ContextTierService( + settings=None, + event_bus=_NullBus(), + ) + frag = _make_frag("evt-bench", ContextTier.COLD) + self.service.store(frag) + + def time_promote_with_event(self) -> None: + self.service.promote("evt-bench") diff --git a/features/context_tier_runtime.feature b/features/context_tier_runtime.feature new file mode 100644 index 000000000..b9b321ba6 --- /dev/null +++ b/features/context_tier_runtime.feature @@ -0,0 +1,179 @@ +@mock_only +Feature: Context tier runtime promotion, demotion, and eviction logic + As an ACMS developer + I want the ContextTierService to automatically manage fragment lifecycle + So that frequently accessed fragments are promoted, stale fragments are + demoted, and budget limits are enforced without manual intervention + + Background: + Given a context tier service with default settings + + # ----------------------------------------------------------------- + # Auto-promotion on access + # ----------------------------------------------------------------- + + Scenario: Fragment promoted from cold to warm after reaching access threshold + Given a fragment "frag-cold-01" stored in the cold tier with 50 tokens + When I access "frag-cold-01" via get 5 times + Then fragment "frag-cold-01" should be in the warm tier + + Scenario: Fragment promoted from warm to hot after reaching access threshold + Given a fragment "frag-warm-01" stored in the warm tier with 50 tokens + When I access "frag-warm-01" via get 5 times + Then fragment "frag-warm-01" should be in the hot tier + + Scenario: Fragment not promoted before reaching access threshold + Given a fragment "frag-cold-02" stored in the cold tier with 50 tokens + When I access "frag-cold-02" via get 3 times + Then fragment "frag-cold-02" should be in the cold tier + + Scenario: Promoted fragment requires fresh accesses for next promotion + Given a fragment "frag-chain-01" stored in the cold tier with 50 tokens + When I access "frag-chain-01" via get 5 times + Then fragment "frag-chain-01" should be in the warm tier + When I access "frag-chain-01" via get 1 times + Then fragment "frag-chain-01" should be in the warm tier + When I access "frag-chain-01" via get 5 times + Then fragment "frag-chain-01" should be in the hot tier + + # ----------------------------------------------------------------- + # Staleness enforcement + # ----------------------------------------------------------------- + + Scenario: Stale hot fragment demoted to warm on staleness enforcement + Given a fragment "frag-stale-hot" stored in the hot tier with stale timestamp of 25 hours + When I invoke enforce_staleness on the service + Then fragment "frag-stale-hot" should be in the warm tier + + Scenario: Fresh hot fragment not demoted on staleness enforcement + Given a fragment "frag-fresh-hot" stored in the hot tier with fresh timestamp + When I invoke enforce_staleness on the service + Then fragment "frag-fresh-hot" should be in the hot tier + + Scenario: Stale warm fragment demoted to cold on staleness enforcement + Given a fragment "frag-stale-warm" stored in the warm tier with stale timestamp of 25 hours + When I invoke enforce_staleness on the service + Then fragment "frag-stale-warm" should be in the cold tier + + Scenario: Staleness enforcement returns list of demoted fragment IDs + Given a fragment "frag-stale-a" stored in the hot tier with stale timestamp of 25 hours + And a fragment "frag-stale-b" stored in the hot tier with stale timestamp of 30 hours + When I invoke enforce_staleness on the service + Then the staleness result should contain 2 fragment IDs + + # ----------------------------------------------------------------- + # Budget enforcement on store + # ----------------------------------------------------------------- + + Scenario: Storing in hot tier when budget allows does not evict + Given a service with hot tier budget of 200 tokens + And a fragment "frag-a" stored in the hot tier with 50 tokens + When I store fragment "frag-b" in the hot tier with 50 tokens + Then fragment "frag-a" should be in the hot tier + And fragment "frag-b" should be in the hot tier + + Scenario: Storing in hot tier over budget evicts oldest fragment + Given a service with hot tier budget of 100 tokens + And a fragment "frag-old" stored in the hot tier with 50 tokens and old timestamp + And a fragment "frag-new" stored in the hot tier with 50 tokens and recent timestamp + When I store fragment "frag-overflow" in the hot tier with 50 tokens + Then fragment "frag-old" should not be in any tier + And fragment "frag-new" should be in the hot tier + And fragment "frag-overflow" should be in the hot tier + + Scenario: Oversized fragment redirected to warm tier instead of hot + Given a service with hot tier budget of 100 tokens + When I store fragment "frag-oversized" in the hot tier with 200 tokens + Then fragment "frag-oversized" should be in the warm tier + + Scenario: Promotion to hot falls back to warm when budget evicts promoted fragment + Given a context tier service with an event bus and hot budget of 100 tokens + And a fragment "frag-resident" stored in the hot tier with 80 tokens and recent timestamp + And a fragment "frag-to-promote" stored in the warm tier with 80 tokens and old timestamp + When I promote fragment "frag-to-promote" + Then fragment "frag-to-promote" should be in the warm tier + And a TIER_PROMOTED event should have been emitted for "frag-to-promote" + + # ----------------------------------------------------------------- + # Event emission + # ----------------------------------------------------------------- + + Scenario: Promotion emits TIER_PROMOTED event with correct tier details + Given a context tier service with an event bus + And a fragment "frag-evt-01" stored in the cold tier with 50 tokens + When I promote fragment "frag-evt-01" + Then a TIER_PROMOTED event should have been emitted for "frag-evt-01" + And the last TIER_PROMOTED event should have from_tier "cold" and to_tier "warm" + + Scenario: Demotion emits TIER_DEMOTED event with correct tier details + Given a context tier service with an event bus + And a fragment "frag-evt-02" stored in the hot tier with 50 tokens + When I demote fragment "frag-evt-02" + Then a TIER_DEMOTED event should have been emitted for "frag-evt-02" + And the last TIER_DEMOTED event should have from_tier "hot" and to_tier "warm" + + Scenario: Budget eviction emits TIER_EVICTED event + Given a context tier service with an event bus and hot budget of 100 tokens + And a fragment "frag-evt-03" stored in the hot tier with 60 tokens and old timestamp + When I store fragment "frag-evt-04" in the hot tier with 60 tokens + Then a TIER_EVICTED event should have been emitted for "frag-evt-03" + + Scenario: Explicit evict_lru emits TIER_EVICTED events + Given a context tier service with an event bus + And a fragment "frag-evict-01" stored in the hot tier with 50 tokens and old timestamp + And a fragment "frag-evict-02" stored in the hot tier with 50 tokens and recent timestamp + When I evict 1 LRU fragment from the hot tier + Then a TIER_EVICTED event should have been emitted for "frag-evict-01" + + Scenario: Oversized fragment redirect emits TIER_DEMOTED event + Given a context tier service with an event bus and hot budget of 100 tokens + When I store fragment "frag-oversized-evt" in the hot tier with 200 tokens + Then fragment "frag-oversized-evt" should be in the warm tier + And a TIER_DEMOTED event should have been emitted for "frag-oversized-evt" + + Scenario: Event emission failure does not break tier operation + Given a context tier service with a failing event bus + And a fragment "frag-evt-safe" stored in the cold tier with 50 tokens + When I promote fragment "frag-evt-safe" + Then fragment "frag-evt-safe" should be in the warm tier + + # ----------------------------------------------------------------- + # Edge cases + # ----------------------------------------------------------------- + + Scenario: Getting a nonexistent fragment returns nothing + When I access "no-such-fragment" via get 1 times + Then fragment "no-such-fragment" should not be in any tier + + Scenario: Promoting a hot-tier fragment has no effect + Given a fragment "frag-already-hot" stored in the hot tier with 50 tokens + When I promote fragment "frag-already-hot" + Then fragment "frag-already-hot" should be in the hot tier + + Scenario: Demoting a cold-tier fragment has no effect + Given a fragment "frag-already-cold" stored in the cold tier with 50 tokens + When I demote fragment "frag-already-cold" + Then fragment "frag-already-cold" should be in the cold tier + + Scenario: Staleness enforcement with mixed stale hot and warm fragments + Given a context tier service with default settings + And a fragment "mixed-hot" stored in the hot tier with stale timestamp of 25 hours + And a fragment "mixed-warm" stored in the warm tier with stale timestamp of 25 hours + When I invoke enforce_staleness on the service + Then fragment "mixed-hot" should be in the warm tier + And fragment "mixed-warm" should be in the cold tier + And the staleness result should contain 2 fragment IDs + + Scenario: Demoted fragment requires fresh accesses before re-promotion + Given a context tier service with default settings + And a fragment "frag-pop" stored in the hot tier with 50 tokens + When I access "frag-pop" via get 10 times + And I invoke enforce_staleness on the service with a hot TTL of 0 hours + And I access "frag-pop" via get 1 times + Then fragment "frag-pop" should be in the warm tier + + Scenario: Promotion threshold of 1 promotes on first access without chain promotion + Given a context tier service with promotion threshold 1 + And a fragment "frag-thresh" stored in the cold tier with 50 tokens + When I access "frag-thresh" via get 1 times + Then fragment "frag-thresh" should be in the warm tier diff --git a/features/steps/context_tier_runtime_steps.py b/features/steps/context_tier_runtime_steps.py new file mode 100644 index 000000000..50083398b --- /dev/null +++ b/features/steps/context_tier_runtime_steps.py @@ -0,0 +1,367 @@ +"""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 diff --git a/features/tdd_context_tier_runtime.feature b/features/tdd_context_tier_runtime.feature index ef022b452..335566c02 100644 --- a/features/tdd_context_tier_runtime.feature +++ b/features/tdd_context_tier_runtime.feature @@ -1,4 +1,4 @@ -@tdd_expected_fail @tdd_bug @tdd_bug_821 @mock_only +@tdd_bug @tdd_bug_821 @mock_only Feature: TDD Bug #821 — context tier service has data models but no runtime logic As a developer I want to verify that ContextTierService automatically promotes, demotes, diff --git a/robot/context_tier_runtime.robot b/robot/context_tier_runtime.robot new file mode 100644 index 000000000..42dff8637 --- /dev/null +++ b/robot/context_tier_runtime.robot @@ -0,0 +1,43 @@ +*** Settings *** +Documentation Integration tests for ACMS Context Tier runtime logic (issue #821) +... Verifies auto-promotion, staleness enforcement, budget eviction, +... and tier transition event emission. +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER} ${CURDIR}/helper_context_tier_runtime.py + +*** Test Cases *** +Auto Promotion On Repeated Access + [Documentation] Accessing a cold-tier fragment 5 times auto-promotes it + ${result}= Run Process ${PYTHON} ${HELPER} auto-promote cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} context-tier-runtime-auto-promote-ok + +Staleness Enforcement Demotes Stale Fragments + [Documentation] enforce_staleness() demotes hot fragments older than TTL + ${result}= Run Process ${PYTHON} ${HELPER} staleness cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} context-tier-runtime-staleness-ok + +Budget Enforcement Evicts On Overflow + [Documentation] Storing beyond hot-tier token budget evicts LRU fragment + ${result}= Run Process ${PYTHON} ${HELPER} budget-evict cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} context-tier-runtime-budget-evict-ok + +Tier Transition Event Emission + [Documentation] Promotion/demotion/eviction emits DomainEvent via EventBus + ${result}= Run Process ${PYTHON} ${HELPER} event-emission cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} context-tier-runtime-event-emission-ok diff --git a/robot/helper_context_tier_runtime.py b/robot/helper_context_tier_runtime.py new file mode 100644 index 000000000..78158df81 --- /dev/null +++ b/robot/helper_context_tier_runtime.py @@ -0,0 +1,158 @@ +"""Robot Framework helper for context tier runtime integration tests. + +Each command exercises the runtime logic added by issue #821: +auto-promotion, staleness enforcement, and budget enforcement. + +Usage:: + + python helper_context_tier_runtime.py + +Commands: + auto-promote -- Verify auto-promotion on repeated access. + staleness -- Verify enforce_staleness() demotes stale fragments. + budget-evict -- Verify budget enforcement on store. + event-emission -- Verify tier transition events are emitted. +""" + +from __future__ import annotations + +import sys +from collections.abc import Callable +from datetime import UTC, datetime, timedelta +from pathlib import Path + +# Ensure source tree is importable +_SRC = str(Path(__file__).resolve().parents[1] / "src") +if _SRC not in sys.path: + sys.path.insert(0, _SRC) + +from cleveragents.application.services.context_tiers import ( # noqa: E402 + ContextTierService, +) +from cleveragents.domain.models.acms.tiers import ( # noqa: E402 + ContextTier, + TierBudget, + TieredFragment, +) +from cleveragents.infrastructure.events.models import DomainEvent # noqa: E402 +from cleveragents.infrastructure.events.types import EventType # noqa: E402 + + +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 + + +def _make_frag( + fid: str, + tier: ContextTier, + tokens: int = 50, + last_accessed: datetime | None = None, +) -> TieredFragment: + frag = TieredFragment( + fragment_id=fid, + content="test", + tier=tier, + token_count=tokens, + project_name="test", + ) + if last_accessed is not None: + frag.last_accessed = last_accessed + return frag + + +def cmd_auto_promote() -> None: + svc = ContextTierService(settings=None) + frag = _make_frag("frag-01", ContextTier.COLD) + svc.store(frag) + + for _ in range(5): + svc.get("frag-01") + + result = svc._find_fragment("frag-01") + assert result is not None, "Fragment disappeared" + assert result.tier == ContextTier.WARM, ( + f"Expected WARM (not HOT — counter reset prevents chain promotion)" + f" after 5 accesses from cold, got {result.tier}" + ) + print("context-tier-runtime-auto-promote-ok") + + +def cmd_staleness() -> None: + svc = ContextTierService(settings=None) + stale = datetime.now(tz=UTC) - timedelta(hours=25) + frag = _make_frag("stale-01", ContextTier.HOT, last_accessed=stale) + svc.store(frag) + + demoted = svc.enforce_staleness() + assert "stale-01" in demoted, f"Expected stale-01 in demoted list: {demoted}" + + result = svc._find_fragment("stale-01") + assert result is not None, "Fragment disappeared" + assert result.tier == ContextTier.WARM, ( + f"Expected WARM after staleness demotion, got {result.tier}" + ) + print("context-tier-runtime-staleness-ok") + + +def cmd_budget_evict() -> None: + svc = ContextTierService(settings=None) + svc._budget = TierBudget(max_tokens_hot=100) + + old = datetime.now(tz=UTC) - timedelta(minutes=10) + svc.store(_make_frag("old-01", ContextTier.HOT, tokens=50, last_accessed=old)) + svc.store(_make_frag("new-01", ContextTier.HOT, tokens=50)) + svc.store(_make_frag("overflow-01", ContextTier.HOT, tokens=50)) + + total = sum(f.token_count for f in svc._hot.values()) + assert total <= 100, f"Hot tier over budget: {total} > 100" + + old_frag = svc._find_fragment("old-01") + assert old_frag is None, "old-01 should have been evicted" + print("context-tier-runtime-budget-evict-ok") + + +def cmd_event_emission() -> None: + collector = _EventCollector() + svc = ContextTierService(settings=None, event_bus=collector) + + svc.store(_make_frag("evt-01", ContextTier.COLD)) + svc.promote("evt-01") + + promoted = [e for e in collector.events if e.event_type == EventType.TIER_PROMOTED] + assert len(promoted) >= 1, f"No TIER_PROMOTED events: {collector.events}" + assert promoted[0].details["fragment_id"] == "evt-01" + print("context-tier-runtime-event-emission-ok") + + +_COMMANDS: dict[str, Callable[[], None]] = { + "auto-promote": cmd_auto_promote, + "staleness": cmd_staleness, + "budget-evict": cmd_budget_evict, + "event-emission": cmd_event_emission, +} + + +def main() -> None: + if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS: + valid = ", ".join(sorted(_COMMANDS.keys())) + print(f"Usage: {sys.argv[0]} <{valid}>", file=sys.stderr) + sys.exit(1) + cmd = _COMMANDS[sys.argv[1]] + cmd() + + +if __name__ == "__main__": + main() diff --git a/robot/tdd_context_tier_runtime.robot b/robot/tdd_context_tier_runtime.robot index 4df2dc225..cc5b66999 100644 --- a/robot/tdd_context_tier_runtime.robot +++ b/robot/tdd_context_tier_runtime.robot @@ -17,7 +17,7 @@ ${HELPER} ${CURDIR}/helper_tdd_context_tier_runtime.py TDD Promotion On Repeated Access [Documentation] Verify that accessing a cold-tier fragment repeatedly via get() ... auto-promotes it to warm or hot tier. - [Tags] tdd_bug tdd_bug_821 tdd_expected_fail + [Tags] tdd_bug tdd_bug_821 ${result}= Run Process ${PYTHON} ${HELPER} promote-on-access cwd=${WORKSPACE} timeout=30s Log ${result.stdout} Log ${result.stderr} @@ -27,7 +27,7 @@ TDD Promotion On Repeated Access TDD Demotion On Staleness [Documentation] Verify that a stale hot-tier fragment is automatically demoted ... to warm or cold tier by a staleness enforcement runtime. - [Tags] tdd_bug tdd_bug_821 tdd_expected_fail + [Tags] tdd_bug tdd_bug_821 ${result}= Run Process ${PYTHON} ${HELPER} demote-on-staleness cwd=${WORKSPACE} timeout=30s Log ${result.stdout} Log ${result.stderr} @@ -37,7 +37,7 @@ TDD Demotion On Staleness TDD Eviction On Budget Overflow [Documentation] Verify that storing beyond the hot tier token budget triggers ... automatic LRU eviction of the oldest fragment. - [Tags] tdd_bug tdd_bug_821 tdd_expected_fail + [Tags] tdd_bug tdd_bug_821 ${result}= Run Process ${PYTHON} ${HELPER} evict-on-overflow cwd=${WORKSPACE} timeout=30s Log ${result.stdout} Log ${result.stderr} diff --git a/src/cleveragents/application/container.py b/src/cleveragents/application/container.py index 360c33cb7..40989d4b9 100644 --- a/src/cleveragents/application/container.py +++ b/src/cleveragents/application/container.py @@ -634,9 +634,11 @@ class Container(containers.DeclarativeContainer): ) # Context Tier Service - Singleton so all callers share tier state + # event_bus injected for tier transition event emission (#821) context_tier_service = providers.Singleton( ContextTierService, settings=settings, + event_bus=event_bus, ) # Decomposition Service - Factory (uses DecisionService + Settings) diff --git a/src/cleveragents/application/services/context_tiers.py b/src/cleveragents/application/services/context_tiers.py index 008d369d4..88fb87afc 100644 --- a/src/cleveragents/application/services/context_tiers.py +++ b/src/cleveragents/application/services/context_tiers.py @@ -1,20 +1,26 @@ """Context Tier Service for hot/warm/cold fragment management. Core tier lifecycle: store, get, promote, demote, evict, metrics. +Runtime logic (auto-promotion, staleness enforcement, budget-based +eviction, event emission) is provided by ``TierRuntimeMixin`` in +``tier_runtime.py``. + Scope enforcement methods live in ``scoped_tiers.py`` and are mixed in via ``ScopedTierMixin``. Based on ``docs/specification.md`` ACMS tier sections, issue #208, -and issue #193 (scoped backend view filtering). +issue #193 (scoped backend view filtering), and issue #821 +(runtime promotion/demotion/eviction). """ from __future__ import annotations -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta import structlog from cleveragents.application.services.scoped_tiers import ScopedTierMixin +from cleveragents.application.services.tier_runtime import TierRuntimeMixin from cleveragents.config.settings import Settings from cleveragents.domain.models.acms.scoped_view import ( ScopedBackendView, @@ -27,6 +33,8 @@ from cleveragents.domain.models.acms.tiers import ( TieredFragment, TierMetrics, ) +from cleveragents.infrastructure.events.protocol import EventBus +from cleveragents.infrastructure.events.types import EventType logger = structlog.get_logger(__name__) @@ -38,21 +46,69 @@ _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 -class ContextTierService(ScopedTierMixin): +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``. + + .. 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. """ - def __init__(self, settings: Settings | None = None) -> None: + def __init__( + self, + settings: Settings | None = None, + event_bus: EventBus | None = None, + ) -> None: budget = _budget_from_settings(settings) self._budget = budget + self._event_bus = event_bus + + # --- 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) # --- tier stores --------------------------------------------------- self._hot: dict[str, TieredFragment] = {} @@ -73,7 +129,14 @@ class ContextTierService(ScopedTierMixin): """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. + 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. @@ -88,7 +151,27 @@ class ContextTierService(ScopedTierMixin): self._remove_from_all(fragment.fragment_id) if fragment.tier == ContextTier.HOT: - self._hot[fragment.fragment_id] = fragment + 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: @@ -101,6 +184,10 @@ class ContextTierService(ScopedTierMixin): def get(self, fragment_id: str) -> TieredFragment | None: """Retrieve a fragment from any tier, updating access metadata. + When a fragment's ``access_count`` reaches the promotion + threshold, it is automatically promoted one tier up + (cold → warm, warm → hot). + Returns ``None`` when the fragment is not found. """ if fragment_id in self._hot: @@ -113,12 +200,16 @@ class ContextTierService(ScopedTierMixin): if fragment_id in self._warm: self._warm_hit += 1 frag = self._warm[fragment_id] - return self._touch(frag) + touched = self._touch(frag) + self._maybe_auto_promote(fragment_id, touched) + return self._re_fetch_after_promotion(fragment_id, touched) self._warm_miss += 1 if fragment_id in self._cold: - return self._touch(self._cold[fragment_id]) + 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 @@ -199,18 +290,52 @@ class ContextTierService(ScopedTierMixin): """Promote a fragment one tier up (cold→warm or warm→hot). Returns the promoted fragment or ``None`` if not found or - already in the hot tier. + already in the hot tier. Emits a ``TIER_PROMOTED`` event. + + Raises: + ValueError: If *fragment_id* is empty. """ + 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}, + ) + 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 @@ -220,19 +345,48 @@ class ContextTierService(ScopedTierMixin): When demoting to cold tier, the summarisation hook is applied. Returns the demoted fragment or ``None`` if not found or - already in the cold tier. + 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}) + 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}) + 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 @@ -264,6 +418,12 @@ class ContextTierService(ScopedTierMixin): 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 # ------------------------------------------------------------------ @@ -369,3 +529,52 @@ def _budget_from_settings(settings: Settings | None) -> TierBudget: 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/tier_runtime.py b/src/cleveragents/application/services/tier_runtime.py new file mode 100644 index 000000000..035d090b3 --- /dev/null +++ b/src/cleveragents/application/services/tier_runtime.py @@ -0,0 +1,251 @@ +"""Runtime policy mixin for the Context Tier Service. + +Extracted from ``context_tiers.py`` to keep modules under the 500-line +guideline. Provides auto-promotion, staleness enforcement, budget +enforcement, and event emission methods that are mixed into +``ContextTierService``. + +All methods assume access to ``_hot``, ``_warm``, ``_cold`` tier stores, +``_budget``, ``_promotion_threshold``, ``_hot_ttl``, ``_warm_ttl``, +``_event_bus``, and the ``promote``, ``demote``, ``_find_fragment`` +helpers defined on the host class. + +Based on ``docs/specification.md`` ACMS tier sections and issue #821 +(runtime promotion/demotion/eviction). +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +import structlog + +from cleveragents.domain.models.acms.tiers import ( + ContextTier, + TierBudget, + TieredFragment, +) +from cleveragents.infrastructure.events.models import DomainEvent +from cleveragents.infrastructure.events.protocol import EventBus +from cleveragents.infrastructure.events.types import EventType + +logger = structlog.get_logger(__name__) + + +class TierRuntimeMixin: + """Mixin providing runtime tier management for ``ContextTierService``. + + Requires the host class to provide: + + - ``_hot``, ``_warm``, ``_cold``: ``dict[str, TieredFragment]`` + - ``_budget``: ``TierBudget`` + - ``_promotion_threshold``: ``int`` + - ``_hot_ttl``: ``timedelta`` + - ``_warm_ttl``: ``timedelta`` + - ``_event_bus``: ``EventBus | None`` + - ``promote(fragment_id: str) -> TieredFragment | None`` + - ``demote(fragment_id: str) -> TieredFragment | None`` + - ``_find_fragment(fragment_id: str) -> TieredFragment | None`` + """ + + # Declared for type checkers; actual attrs live on the host class. + _hot: dict[str, TieredFragment] + _warm: dict[str, TieredFragment] + _cold: dict[str, TieredFragment] + _budget: TierBudget + _promotion_threshold: int + _hot_ttl: timedelta + _warm_ttl: timedelta + _event_bus: EventBus | None + + def promote(self, fragment_id: str) -> TieredFragment | None: ... + def demote(self, fragment_id: str) -> TieredFragment | None: ... + def _find_fragment(self, fragment_id: str) -> TieredFragment | None: ... + + # ------------------------------------------------------------------ + # Staleness enforcement (runtime — issue #821) + # ------------------------------------------------------------------ + + def enforce_staleness(self) -> list[str]: + """Enforce time-based demotion across all tiers. + + Inspects every fragment's ``last_accessed`` timestamp: + + * Hot fragments older than ``hot_ttl`` → demoted to warm. + * Warm fragments older than ``warm_ttl`` → demoted to cold. + + Each pass only considers fragments that were already present + in that tier before this method was called — fragments freshly + demoted from a higher tier are not immediately re-demoted. + + .. note:: + + The specification's tier table describes hot-tier retention + as *"Until resource removed"*. The ``hot_ttl`` controls + tier **placement** (demotion to warm), not data retention + (deletion). Demoted fragments continue to exist in lower + tiers, consistent with the spec's retention semantics. + + Returns a list of fragment IDs that were demoted. + """ + 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()) + + # 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, + ) + + return demoted_ids + + # ------------------------------------------------------------------ + # Auto-promotion (runtime — issue #821) + # ------------------------------------------------------------------ + + def _maybe_auto_promote( + self, + fragment_id: str, + fragment: TieredFragment, + ) -> None: + """Promote *fragment* if its access count meets the threshold. + + Called by :meth:`get` after touching the fragment. Only + promotes one tier at a time (cold → warm, warm → hot). + The access counter is reset after each successful promotion + so the fragment must accumulate fresh accesses before it can + be promoted again. + """ + if fragment.access_count >= self._promotion_threshold: + promoted = self.promote(fragment_id) + if promoted is not None: + # Reset counter so the fragment must accumulate + # fresh accesses before the next promotion. + promoted.access_count = 0 + logger.info( + "tier.auto_promoted", + fragment_id=fragment_id, + access_count=fragment.access_count, + new_tier=promoted.tier.value, + ) + + def _re_fetch_after_promotion( + self, + fragment_id: str, + fallback: TieredFragment, + ) -> TieredFragment: + """Return the current state of *fragment_id* after possible promotion. + + After auto-promotion the fragment may have moved to a different + store, so we re-look it up. Falls back to *fallback* if the + fragment is not found (should not happen). + """ + found = self._find_fragment(fragment_id) + return found if found is not None else fallback + + # ------------------------------------------------------------------ + # Budget enforcement (runtime — issue #821) + # ------------------------------------------------------------------ + + def _enforce_hot_budget(self) -> None: + """Evict LRU hot-tier fragments when the token budget is exceeded. + + Called automatically by :meth:`store` and :meth:`promote` after + inserting a fragment into the hot tier. + + Complexity: **O(n + n*k)** where *n* is the number of hot-tier + fragments and *k* is the number of evictions. Token tracking + uses incremental subtraction (O(1) per eviction), but each + eviction selects the LRU fragment via ``min()`` over the + remaining entries (O(n) per eviction). + """ + budget_tokens = self._budget.max_tokens_hot + total_tokens = sum(f.token_count for f in self._hot.values()) + while total_tokens > budget_tokens and self._hot: + # Evict the single oldest fragment + oldest_id = min( + self._hot.keys(), + key=lambda fid: self._hot[fid].last_accessed, + ) + evicted_tokens = self._hot[oldest_id].token_count + del self._hot[oldest_id] + total_tokens -= evicted_tokens + self._emit_tier_event( + EventType.TIER_EVICTED, + oldest_id, + from_tier=ContextTier.HOT, + to_tier=None, + ) + logger.info( + "tier.budget_evicted", + fragment_id=oldest_id, + hot_tokens=total_tokens + evicted_tokens, + budget=budget_tokens, + ) + + # ------------------------------------------------------------------ + # Event emission (runtime — issue #821) + # ------------------------------------------------------------------ + + def _emit_tier_event( + self, + event_type: EventType, + fragment_id: str, + from_tier: ContextTier, + to_tier: ContextTier | None, + ) -> None: + """Emit a tier transition event via the event bus (if available). + + Emission is best-effort: a failing event bus does not break + the tier operation in progress. + """ + if self._event_bus is None: + return + try: + details: dict[str, str | None] = { + "fragment_id": fragment_id, + "from_tier": from_tier.value, + "to_tier": to_tier.value if to_tier is not None else None, + } + event = DomainEvent( + event_type=event_type, + details=details, + ) + self._event_bus.emit(event) + except Exception: + logger.warning( + "tier.event_emission_failed", + event_type=event_type.value, + fragment_id=fragment_id, + exc_info=True, + ) diff --git a/src/cleveragents/config/settings.py b/src/cleveragents/config/settings.py index 41175ff9b..e361b6216 100644 --- a/src/cleveragents/config/settings.py +++ b/src/cleveragents/config/settings.py @@ -277,6 +277,32 @@ class Settings(BaseSettings): description="Maximum fragments in the cold context tier.", ) + # Context tier runtime settings (ACMS #821) + context_tier_promotion_threshold: int = Field( + default=5, + ge=1, + validation_alias=AliasChoices("CLEVERAGENTS_CONTEXT_TIER_PROMOTION_THRESHOLD"), + description="Number of accesses before a fragment is auto-promoted.", + ) + context_tier_hot_ttl_hours: int = Field( + default=24, + ge=1, + validation_alias=AliasChoices( + "CLEVERAGENTS_CONTEXT_TIER_HOT_TTL_HOURS", + "CLEVERAGENTS_CTX_HOT_HOURS", + ), + description="Hours without access before hot fragment is demoted to warm.", + ) + context_tier_warm_ttl_hours: int = Field( + default=24, + ge=1, + validation_alias=AliasChoices( + "CLEVERAGENTS_CONTEXT_TIER_WARM_TTL_HOURS", + "CLEVERAGENTS_CTX_WARM_HOURS", + ), + description="Hours without access before warm fragment is demoted to cold.", + ) + # Vector store configuration vector_store_enabled: bool = Field( default=False, diff --git a/src/cleveragents/infrastructure/events/types.py b/src/cleveragents/infrastructure/events/types.py index 6b36f90bb..72ec21b4a 100644 --- a/src/cleveragents/infrastructure/events/types.py +++ b/src/cleveragents/infrastructure/events/types.py @@ -82,6 +82,11 @@ class EventType(StrEnum): CONTEXT_BUILT = "context.built" CONTEXT_QUERY_EXECUTED = "context.query_executed" + # --- Context tier lifecycle --- + TIER_PROMOTED = "tier.promoted" + TIER_DEMOTED = "tier.demoted" + TIER_EVICTED = "tier.evicted" + # --- Validation --- VALIDATION_STARTED = "validation.started" VALIDATION_PASSED = "validation.passed"