test(acms): TDD failing tests for context tier runtime logic (bug #821) (#1058)

## Summary

TDD expected-fail tests proving bug #821 exists: `ContextTierService` has data models for hot/warm/cold tiers but **no runtime logic** for automatic promotion, demotion, or eviction.

- **Promotion on access**: Accessing a cold-tier fragment repeatedly via `get()` does NOT auto-promote it — `get()` updates `access_count`/`last_accessed` but never calls `promote()`
- **Demotion on staleness**: No staleness enforcement method exists — tried `enforce_staleness()`, `apply_tier_policy()`, `tick()`, etc. — none are implemented
- **Eviction on budget overflow**: `store()` does NOT enforce `TierBudget.max_tokens_hot` — the hot tier grows without bound

### Files Added
| File | Purpose |
|------|---------|
| `features/tdd_context_tier_runtime.feature` | 3 Behave scenarios tagged `@tdd_expected_fail @tdd_bug @tdd_bug_821 @mock_only` |
| `features/steps/tdd_context_tier_runtime_steps.py` | Type-annotated step definitions exercising real `ContextTierService` |
| `robot/tdd_context_tier_runtime.robot` | 3 Robot Framework integration tests tagged `tdd_expected_fail` |
| `robot/helper_tdd_context_tier_runtime.py` | Helper script for Robot tests with 3 subcommands |

### Verification
- `nox -s lint` — passed
- `nox -s typecheck` — passed (0 errors)
- `nox -s unit_tests -- features/tdd_context_tier_runtime.feature` — **3 scenarios passed** (all assertions fail as expected, `@tdd_expected_fail` inverts to CI pass)

ISSUES CLOSED: #840

Reviewed-on: #1058
Co-authored-by: Brent Edwards <brent.edwards@cleverthis.com>
Co-committed-by: Brent Edwards <brent.edwards@cleverthis.com>

features/steps/tdd_context_tier_runtime_steps.py

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

features/tdd_context_tier_runtime.feature

@@ -0,0 +1,39 @@
+@tdd_expected_fail @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,
+  and evicts fragments based on access patterns, staleness, and budget limits
+  So that the bug is captured and will be caught by a regression test
+
+  ContextTierService has well-defined data models for hot/warm/cold tiers
+  (ContextTier, TieredFragment, TierBudget) and manual promote()/demote()/
+  evict_lru() methods, but NO automatic runtime logic:
+
+  - get() touches access metadata but never auto-promotes a frequently
+    accessed cold/warm fragment to a higher tier.
+  - There is no staleness enforcement: no method checks last_accessed
+    timestamps and demotes stale hot fragments to warm/cold.
+  - store() does not enforce budget limits: storing beyond
+    max_tokens_hot does not trigger automatic eviction.
+
+  These tests assert the expected runtime behaviour and will FAIL until
+  the bug is fixed.  The @tdd_expected_fail tag inverts the result so
+  CI passes.
+
+  Scenario: Promotion on repeated access moves fragment to a higher tier
+    Given a context tier service with default budget
+    And a fragment stored in the cold tier
+    When I access the fragment 5 times via get
+    Then the fragment should have been promoted to warm or hot tier
+
+  Scenario: Demotion on staleness moves fragment to a lower tier
+    Given a context tier service with default budget
+    And a fragment stored in the hot tier with a stale last_accessed timestamp
+    When I invoke the staleness enforcement runtime
+    Then the fragment should have been demoted to warm or cold tier
+
+  Scenario: Eviction on hot tier budget overflow removes oldest fragment
+    Given a context tier service with a small hot tier budget of 100 tokens
+    And the hot tier is filled to its token budget limit
+    When I store one more fragment in the hot tier
+    Then the oldest fragment should have been evicted from the hot tier

robot/helper_tdd_context_tier_runtime.py

@@ -0,0 +1,232 @@
+"""Helper script for tdd_context_tier_runtime.robot smoke tests.
+
+Each subcommand exercises ``ContextTierService`` to reproduce bug #821.
+The helper reports the **real** outcome: it exits 0 and prints the sentinel
+when the expected behaviour is observed (bug fixed), and exits 1 when the
+bug is still present.  The ``tdd_expected_fail_listener`` on the Robot side
+handles pass/fail inversion while the bug remains open.
+"""
+
+from __future__ import annotations
+
+import sys
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import NoReturn
+
+# Ensure local source tree is importable.
+_ROOT: Path = Path(__file__).resolve().parents[1]
+_SRC: str = str(_ROOT / "src")
+_ROBOT: str = str(_ROOT / "robot")
+for _p in (_SRC, _ROBOT):
+    if _p not in sys.path:
+        sys.path.insert(0, _p)
+
+from cleveragents.application.services.context_tiers import (  # noqa: E402
+    ContextTierService,
+)
+from cleveragents.domain.models.acms.tiers import (  # noqa: E402
+    ContextTier,
+    TierBudget,
+    TieredFragment,
+)
+
+
+def _fail(msg: str) -> NoReturn:
+    """Print error message to stderr and exit with code 1."""
+    print(msg, file=sys.stderr)
+    sys.exit(1)
+
+
+def _make_fragment(
+    fragment_id: str,
+    tier: ContextTier,
+    token_count: int = 50,
+    content: str = "test content",
+    last_accessed: datetime | None = None,
+) -> TieredFragment:
+    """Create a ``TieredFragment`` with the given properties."""
+    kwargs: dict[str, object] = {
+        "fragment_id": fragment_id,
+        "content": content,
+        "tier": tier,
+        "token_count": token_count,
+        "project_name": "test-project",
+    }
+    if last_accessed is not None:
+        kwargs["last_accessed"] = last_accessed
+    return TieredFragment(**kwargs)  # type: ignore[arg-type]
+
+
+# ---------------------------------------------------------------------------
+# Subcommands
+# ---------------------------------------------------------------------------
+
+
+def _promote_on_access() -> None:
+    """Verify that accessing a cold-tier fragment repeatedly auto-promotes it.
+
+    Bug #821: ``get()`` increments ``access_count`` and updates
+    ``last_accessed`` but never calls ``promote()`` or any equivalent
+    runtime logic.  The fragment stays in the cold tier forever.
+    """
+    service: ContextTierService = ContextTierService(settings=None)
+    frag: TieredFragment = _make_fragment(
+        fragment_id="frag-promote-001",
+        tier=ContextTier.COLD,
+        token_count=50,
+    )
+    service.store(frag)
+
+    # Access the fragment 5 times
+    for _ in range(5):
+        result: TieredFragment | None = service.get("frag-promote-001")
+        if result is None:
+            _fail("Fragment disappeared during repeated access")
+
+    # Check if the fragment was promoted
+    final: TieredFragment | None = service.get("frag-promote-001")
+    if final is None:
+        _fail("Fragment not found after repeated access")
+
+    if final.tier == ContextTier.COLD:
+        _fail(
+            f"Fragment is still in COLD tier after 5 accesses "
+            f"(access_count={final.access_count}). "
+            f"ContextTierService.get() does not auto-promote fragments "
+            f"based on access patterns (bug #821)."
+        )
+    print("tdd-promote-on-access-ok")
+
+
+def _demote_on_staleness() -> None:
+    """Verify that a stale hot-tier fragment is auto-demoted.
+
+    Bug #821: ContextTierService has no staleness enforcement method.
+    No ``enforce_staleness()``, ``apply_tier_policy()``, ``tick()``,
+    or similar runtime method exists.
+    """
+    service: ContextTierService = ContextTierService(settings=None)
+    stale_time: datetime = datetime.now(tz=UTC) - timedelta(hours=24)
+    frag: TieredFragment = _make_fragment(
+        fragment_id="frag-stale-001",
+        tier=ContextTier.HOT,
+        token_count=100,
+        last_accessed=stale_time,
+    )
+    service.store(frag)
+
+    # Attempt to invoke staleness enforcement
+    invoked: bool = False
+    for method_name in (
+        "enforce_staleness",
+        "apply_tier_policy",
+        "tick",
+        "enforce_demotion_policy",
+        "run_lifecycle",
+        "apply_lifecycle",
+    ):
+        method: object = getattr(service, method_name, None)
+        if callable(method):
+            method()
+            invoked = True
+            break
+
+    if not invoked:
+        _fail(
+            "No staleness enforcement method found on ContextTierService. "
+            "Tried: enforce_staleness(), apply_tier_policy(), tick(), "
+            "enforce_demotion_policy(), run_lifecycle(), apply_lifecycle(). "
+            "ContextTierService has no runtime logic for automatic "
+            "demotion of stale fragments (bug #821)."
+        )
+
+    # Check if the fragment was demoted
+    result: TieredFragment | None = service.get("frag-stale-001")
+    if result is None:
+        _fail("Stale fragment not found after staleness enforcement")
+
+    if result.tier == ContextTier.HOT:
+        _fail(
+            f"Stale fragment is still in HOT tier "
+            f"(last_accessed={result.last_accessed}). "
+            f"Staleness enforcement did not demote it (bug #821)."
+        )
+    print("tdd-demote-on-staleness-ok")
+
+
+def _evict_on_overflow() -> None:
+    """Verify that hot tier budget overflow triggers auto-eviction.
+
+    Bug #821: ``store()`` does not check ``TierBudget.max_tokens_hot``
+    and does not call ``evict_lru()`` when the budget is exceeded.
+    """
+    service: ContextTierService = ContextTierService(settings=None)
+    service._budget = TierBudget(
+        max_tokens_hot=100,
+        max_decisions_warm=500,
+        max_decisions_cold=5000,
+    )
+
+    # Fill hot tier to exactly 100 tokens (2 x 50-token fragments)
+    for i in range(2):
+        frag: TieredFragment = _make_fragment(
+            fragment_id=f"frag-budget-{i:03d}",
+            tier=ContextTier.HOT,
+            token_count=50,
+            content="x" * 50,
+            last_accessed=datetime.now(tz=UTC) - timedelta(minutes=2 - i),
+        )
+        service.store(frag)
+
+    oldest_id: str = "frag-budget-000"
+
+    # Store one more fragment that exceeds the budget
+    overflow: TieredFragment = _make_fragment(
+        fragment_id="frag-overflow-001",
+        tier=ContextTier.HOT,
+        token_count=50,
+        content="overflow",
+    )
+    service.store(overflow)
+
+    # Check hot tier total tokens
+    hot_total: int = sum(f.token_count for f in service._hot.values())
+    if hot_total > 100:
+        _fail(
+            f"Hot tier has {hot_total} tokens but budget is 100 tokens. "
+            f"store() does not enforce TierBudget.max_tokens_hot or "
+            f"trigger automatic LRU eviction (bug #821)."
+        )
+
+    # Check that the oldest fragment was evicted
+    oldest: TieredFragment | None = service.get(oldest_id)
+    if oldest is not None and oldest.tier == ContextTier.HOT:
+        _fail(
+            f"Oldest fragment '{oldest_id}' is still in hot tier after "
+            f"budget overflow. store() does not auto-evict (bug #821)."
+        )
+
+    print("tdd-evict-on-overflow-ok")
+
+
+# ---------------------------------------------------------------------------
+# Dispatcher
+# ---------------------------------------------------------------------------
+
+_COMMANDS: dict[str, Callable[[], None]] = {
+    "promote-on-access": _promote_on_access,
+    "demote-on-staleness": _demote_on_staleness,
+    "evict-on-overflow": _evict_on_overflow,
+}
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
+        print(
+            f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    cmd: Callable[[], None] = _COMMANDS[sys.argv[1]]
+    cmd()

robot/tdd_context_tier_runtime.robot

@@ -0,0 +1,45 @@
+*** Settings ***
+Documentation    TDD Bug #821 — context tier service has data models but no runtime logic
+...              for promotion, demotion, or eviction.  Integration smoke tests verifying
+...              that ContextTierService automatically promotes fragments on repeated access,
+...              demotes stale fragments, and evicts fragments when the hot tier budget
+...              overflows.  Currently the service has manual promote()/demote()/evict_lru()
+...              methods but no automatic runtime behaviour wired into get() or store().
+...              Tests are tagged tdd_expected_fail so CI passes via result inversion.
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_tdd_context_tier_runtime.py
+
+*** Test Cases ***
+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
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    promote-on-access    cwd=${WORKSPACE}    timeout=30s
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    tdd-promote-on-access-ok
+
+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
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    demote-on-staleness    cwd=${WORKSPACE}    timeout=30s
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    tdd-demote-on-staleness-ok
+
+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
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    evict-on-overflow    cwd=${WORKSPACE}    timeout=30s
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    tdd-evict-on-overflow-ok