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

fix(acms): implement context tier runtime promotion/demotion/eviction #1150

Merged
CoreRasurae merged 1 commits from bugfix/m5-context-tier-runtime into master 2026-03-25 19:04:12 +00:00
Conversation 7 Commits 1 Files Changed 13
+1413 -16
13 changed files with 1413 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+18
View File
@@ -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
benchmarks/bench_context_tier_runtime.py
+139
View File
@@ -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")
features/context_tier_runtime.feature
+179
View File
@@ -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
features/steps/context_tier_runtime_steps.py
+367
View File
@@ -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
features/tdd_context_tier_runtime.feature
+1 -1
View File
@@ -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,
robot/context_tier_runtime.robot
+43
View File
@@ -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
robot/helper_context_tier_runtime.py
+158
View File
@@ -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 <command>
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()
robot/tdd_context_tier_runtime.robot
+3 -3
View File
@@ -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}
src/cleveragents/application/container.py
+2
View File
@@ -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)
src/cleveragents/application/services/context_tiers.py
+221 -12
View File
@@ -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"]
src/cleveragents/application/services/tier_runtime.py
+251
View File
@@ -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,
)
src/cleveragents/config/settings.py
+26
View File
@@ -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,
src/cleveragents/infrastructure/events/types.py
+5
View File
@@ -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"