features/acms_hot_storage_tier.feature

@@ -0,0 +1,242 @@
+Feature: ACMS Hot Storage Tier (in-memory LRU cache)
+  As an ACMS developer
+  I want a hot storage tier backed by an in-memory LRU cache
+  So that recently and frequently accessed context entries are served
+  with minimal latency while capacity limits are enforced automatically
+
+  # ---- Construction ----
+
+  Scenario: Create HotStorageTier with no limits
+    Given a HotStorageTier with no capacity limits
+    Then the hot tier entry_count should be 0
+    And the hot storage tier size_bytes should be 0
+    And the hot tier hit_count should be 0
+    And the hot tier miss_count should be 0
+
+  Scenario: Create HotStorageTier with max_entries limit
+    Given a HotStorageTier with max_entries 10
+    Then the hot tier max_entries should be 10
+    And the hot tier max_bytes should be None
+
+  Scenario: Create HotStorageTier with max_bytes limit
+    Given a HotStorageTier with max_bytes 1024
+    Then the hot tier max_bytes should be 1024
+    And the hot tier max_entries should be None
+
+  Scenario: Create HotStorageTier with both limits
+    Given a HotStorageTier with max_entries 5 and max_bytes 512
+    Then the hot tier max_entries should be 5
+    And the hot tier max_bytes should be 512
+
+  Scenario: HotStorageTier rejects max_entries of zero
+    Then creating a HotStorageTier with max_entries 0 should raise ValueError
+
+  Scenario: HotStorageTier rejects negative max_entries
+    Then creating a HotStorageTier with max_entries -1 should raise ValueError
+
+  Scenario: HotStorageTier rejects max_bytes of zero
+    Then creating a HotStorageTier with max_bytes 0 should raise ValueError
+
+  Scenario: HotStorageTier rejects negative max_bytes
+    Then creating a HotStorageTier with max_bytes -1 should raise ValueError
+
+  # ---- Basic put/get ----
+
+  Scenario: Put and get a single entry
+    Given a HotStorageTier with no capacity limits
+    When I put entry "key1" with content "hello world"
+    Then getting "key1" from the hot tier should return "hello world"
+
+  Scenario: Get returns None for missing entry
+    Given a HotStorageTier with no capacity limits
+    Then getting "missing" from the hot tier should return None
+
+  Scenario: Put updates existing entry
+    Given a HotStorageTier with no capacity limits
+    When I put entry "key1" with content "original"
+    And I put entry "key1" with content "updated"
+    Then getting "key1" from the hot tier should return "updated"
+    And the hot tier entry_count should be 1
+
+  Scenario: Put rejects empty entry_id
+    Given a HotStorageTier with no capacity limits
+    Then putting an entry with empty id should raise ValueError
+
+  # ---- Metrics ----
+
+  Scenario: Hit count increments on successful get
+    Given a HotStorageTier with no capacity limits
+    When I put entry "k1" with content "data"
+    And I get "k1" from the hot tier
+    Then the hot tier hit_count should be 1
+    And the hot tier miss_count should be 0
+
+  Scenario: Miss count increments on failed get
+    Given a HotStorageTier with no capacity limits
+    When I get "nonexistent" from the hot tier
+    Then the hot tier hit_count should be 0
+    And the hot tier miss_count should be 1
+
+  Scenario: Hit and miss counts accumulate independently
+    Given a HotStorageTier with no capacity limits
+    When I put entry "k1" with content "data"
+    And I get "k1" from the hot tier
+    And I get "k1" from the hot tier
+    And I get "missing" from the hot tier
+    Then the hot tier hit_count should be 2
+    And the hot tier miss_count should be 1
+
+  Scenario: entry_count reflects current cache size
+    Given a HotStorageTier with no capacity limits
+    When I put entry "a" with content "aaa"
+    And I put entry "b" with content "bbb"
+    And I put entry "c" with content "ccc"
+    Then the hot tier entry_count should be 3
+
+  Scenario: size_bytes reflects UTF-8 encoded content size
+    Given a HotStorageTier with no capacity limits
+    When I put entry "k1" with content "hello"
+    Then the hot storage tier size_bytes should be 5
+
+  Scenario: size_bytes updates when entry is replaced
+    Given a HotStorageTier with no capacity limits
+    When I put entry "k1" with content "hi"
+    And I put entry "k1" with content "hello world"
+    Then the hot storage tier size_bytes should be 11
+
+  # ---- LRU eviction: max_entries ----
+
+  Scenario: LRU eviction triggers when max_entries is exceeded
+    Given a HotStorageTier with max_entries 3
+    When I put entry "e1" with content "first"
+    And I put entry "e2" with content "second"
+    And I put entry "e3" with content "third"
+    And I put entry "e4" with content "fourth"
+    Then the hot tier entry_count should be 3
+    And getting "e1" from the hot tier should return None
+    And getting "e4" from the hot tier should return "fourth"
+
+  Scenario: LRU order is updated on get
+    Given a HotStorageTier with max_entries 2
+    When I put entry "a" with content "aaa"
+    And I put entry "b" with content "bbb"
+    And I get "a" from the hot tier
+    And I put entry "c" with content "ccc"
+    Then getting "a" from the hot tier should return "aaa"
+    And getting "b" from the hot tier should return None
+    And getting "c" from the hot tier should return "ccc"
+
+  Scenario: LRU order is updated on put (update existing)
+    Given a HotStorageTier with max_entries 2
+    When I put entry "a" with content "aaa"
+    And I put entry "b" with content "bbb"
+    And I put entry "a" with content "aaa-updated"
+    And I put entry "c" with content "ccc"
+    Then getting "a" from the hot tier should return "aaa-updated"
+    And getting "b" from the hot tier should return None
+    And getting "c" from the hot tier should return "ccc"
+
+  # ---- LRU eviction: max_bytes ----
+
+  Scenario: LRU eviction triggers when max_bytes is exceeded
+    Given a HotStorageTier with max_bytes 10
+    When I put entry "k1" with content "12345"
+    And I put entry "k2" with content "67890"
+    And I put entry "k3" with content "abcde"
+    Then the hot storage tier size_bytes should be at most 10
+    And getting "k1" from the hot tier should return None
+
+  Scenario: Single entry larger than max_bytes is evicted immediately
+    Given a HotStorageTier with max_bytes 3
+    When I put entry "big" with content "hello"
+    Then the hot tier entry_count should be 0
+    And the hot storage tier size_bytes should be 0
+
+  # ---- Eviction callback (warm-tier demotion) ----
+
+  Scenario: on_evict callback is called with evicted entry id
+    Given a HotStorageTier with max_entries 2 and an eviction recorder
+    When I put entry "x1" with content "content1"
+    And I put entry "x2" with content "content2"
+    And I put entry "x3" with content "content3"
+    Then the eviction recorder should have recorded entry id "x1"
+
+  Scenario: on_evict callback receives correct content for evicted entry
+    Given a HotStorageTier with max_entries 1 and an eviction recorder
+    When I put entry "first" with content "first-content"
+    And I put entry "second" with content "second-content"
+    Then the eviction recorder should have recorded entry "first" with content "first-content"
+
+  Scenario: on_evict callback error does not interrupt cache operation
+    Given a HotStorageTier with max_entries 1 and a failing eviction callback
+    When I put entry "a" with content "aaa"
+    And I put entry "b" with content "bbb"
+    Then the hot tier entry_count should be 1
+    And getting "b" from the hot tier should return "bbb"
+
+  Scenario: on_evict is not called when no eviction occurs
+    Given a HotStorageTier with max_entries 5 and an eviction recorder
+    When I put entry "only" with content "data"
+    Then the eviction recorder should have recorded 0 evictions
+
+  # ---- Remove ----
+
+  Scenario: Remove an existing entry
+    Given a HotStorageTier with no capacity limits
+    When I put entry "r1" with content "remove-me"
+    And I remove "r1" from the hot tier
+    Then getting "r1" from the hot tier should return None
+    And the hot tier entry_count should be 0
+
+  Scenario: Remove returns the content of the removed entry
+    Given a HotStorageTier with no capacity limits
+    When I put entry "r2" with content "my-content"
+    And I remove "r2" from the hot tier
+    Then removing "r2" from the hot tier should return "my-content"
+
+  Scenario: Remove returns None for missing entry
+    Given a HotStorageTier with no capacity limits
+    When I remove "nonexistent" from the hot tier
+    Then removing "nonexistent" from the hot tier should return None
+
+  Scenario: Remove updates size_bytes correctly
+    Given a HotStorageTier with no capacity limits
+    When I put entry "s1" with content "hello"
+    And I remove "s1" from the hot tier
+    Then the hot storage tier size_bytes should be 0
+
+  # ---- Clear ----
+
+  Scenario: Clear removes all entries
+    Given a HotStorageTier with no capacity limits
+    When I put entry "c1" with content "aaa"
+    And I put entry "c2" with content "bbb"
+    And I clear the hot tier
+    Then the hot tier entry_count should be 0
+    And the hot storage tier size_bytes should be 0
+
+  Scenario: Clear does not reset hit/miss counters
+    Given a HotStorageTier with no capacity limits
+    When I put entry "c1" with content "aaa"
+    And I get "c1" from the hot tier
+    And I clear the hot tier
+    Then the hot tier hit_count should be 1
+
+  # ---- Thread safety ----
+
+  Scenario: Concurrent puts from multiple threads do not corrupt the cache
+    Given a HotStorageTier with max_entries 50
+    When 10 threads concurrently put 5 entries each into the hot tier
+    Then no exception should have been raised during concurrent puts
+    And the hot tier entry_count should be at most 50
+
+  Scenario: Concurrent gets and puts do not raise RuntimeError
+    Given a HotStorageTier with no capacity limits
+    When 5 threads concurrently put entries and 5 threads concurrently get entries
+    Then no exception should have been raised during concurrent gets and puts
+
+  Scenario: Concurrent evictions do not corrupt size_bytes
+    Given a HotStorageTier with max_bytes 100
+    When 8 threads concurrently put large entries into the hot tier
+    Then no exception should have been raised during concurrent evictions
+    And the hot storage tier size_bytes should be at most 100

features/steps/acms_hot_storage_tier_steps.py

@@ -0,0 +1,375 @@
+"""Step definitions for the ACMS Hot Storage Tier feature."""
+
+from __future__ import annotations
+
+import threading
+from typing import Any
+
+from behave import given, then, when
+
+from cleveragents.acms.storage.hot import HotStorageTier
+
+__all__: list[str] = []
+
+
+# ---------------------------------------------------------------------------
+# Construction
+# ---------------------------------------------------------------------------
+
+
+@given("a HotStorageTier with no capacity limits")
+def step_given_hot_tier_no_limits(context: Any) -> None:
+    context.hot_tier = HotStorageTier()
+
+
+@given("a HotStorageTier with max_entries {n:d}")
+def step_given_hot_tier_max_entries(context: Any, n: int) -> None:
+    context.hot_tier = HotStorageTier(max_entries=n)
+
+
+@given("a HotStorageTier with max_bytes {n:d}")
+def step_given_hot_tier_max_bytes(context: Any, n: int) -> None:
+    context.hot_tier = HotStorageTier(max_bytes=n)
+
+
+@given("a HotStorageTier with max_entries {entries:d} and max_bytes {bts:d}")
+def step_given_hot_tier_both_limits(context: Any, entries: int, bts: int) -> None:
+    context.hot_tier = HotStorageTier(max_entries=entries, max_bytes=bts)
+
+
+@given("a HotStorageTier with max_entries {n:d} and an eviction recorder")
+def step_given_hot_tier_with_recorder(context: Any, n: int) -> None:
+    context.evicted_entries: list[tuple[str, str]] = []
+
+    def recorder(entry_id: str, content: str) -> None:
+        context.evicted_entries.append((entry_id, content))
+
+    context.hot_tier = HotStorageTier(max_entries=n, on_evict=recorder)
+
+
+@given("a HotStorageTier with max_entries {n:d} and a failing eviction callback")
+def step_given_hot_tier_failing_callback(context: Any, n: int) -> None:
+    def failing_callback(entry_id: str, content: str) -> None:
+        raise RuntimeError("Simulated callback failure")
+
+    context.hot_tier = HotStorageTier(max_entries=n, on_evict=failing_callback)
+
+
+# ---------------------------------------------------------------------------
+# Validation errors on construction
+# ---------------------------------------------------------------------------
+
+
+@then("creating a HotStorageTier with max_entries {n:d} should raise ValueError")
+def step_then_max_entries_raises(context: Any, n: int) -> None:
+    try:
+        HotStorageTier(max_entries=n)
+        raise AssertionError("Expected ValueError")
+    except ValueError:
+        pass
+
+
+@then("creating a HotStorageTier with max_bytes {n:d} should raise ValueError")
+def step_then_max_bytes_raises(context: Any, n: int) -> None:
+    try:
+        HotStorageTier(max_bytes=n)
+        raise AssertionError("Expected ValueError")
+    except ValueError:
+        pass
+
+
+# ---------------------------------------------------------------------------
+# Put / Get
+# ---------------------------------------------------------------------------
+
+
+@when('I put entry "{entry_id}" with content "{content}"')
+def step_when_put_entry(context: Any, entry_id: str, content: str) -> None:
+    context.hot_tier.put(entry_id, content)
+
+
+@when('I get "{entry_id}" from the hot tier')
+def step_when_get_entry(context: Any, entry_id: str) -> None:
+    context.last_get_result = context.hot_tier.get(entry_id)
+
+
+@then('getting "{entry_id}" from the hot tier should return "{expected}"')
+def step_then_get_returns(context: Any, entry_id: str, expected: str) -> None:
+    result = context.hot_tier.get(entry_id)
+    assert result == expected, f"Expected {expected!r}, got {result!r}"
+
+
+@then('getting "{entry_id}" from the hot tier should return None')
+def step_then_get_returns_none(context: Any, entry_id: str) -> None:
+    result = context.hot_tier.get(entry_id)
+    assert result is None, f"Expected None, got {result!r}"
+
+
+@then("putting an entry with empty id should raise ValueError")
+def step_then_put_empty_id_raises(context: Any) -> None:
+    try:
+        context.hot_tier.put("", "content")
+        raise AssertionError("Expected ValueError")
+    except ValueError:
+        pass
+
+
+# ---------------------------------------------------------------------------
+# Remove
+# ---------------------------------------------------------------------------
+
+
+@when('I remove "{entry_id}" from the hot tier')
+def step_when_remove_entry(context: Any, entry_id: str) -> None:
+    context.last_remove_result = context.hot_tier.remove(entry_id)
+
+
+@then('removing "{entry_id}" from the hot tier should return "{expected}"')
+def step_then_remove_returns(context: Any, entry_id: str, expected: str) -> None:
+    result = context.last_remove_result
+    assert result == expected, f"Expected {expected!r}, got {result!r}"
+
+
+@then('removing "{entry_id}" from the hot tier should return None')
+def step_then_remove_returns_none(context: Any, entry_id: str) -> None:
+    result = context.last_remove_result
+    assert result is None, f"Expected None, got {result!r}"
+
+
+# ---------------------------------------------------------------------------
+# Clear
+# ---------------------------------------------------------------------------
+
+
+@when("I clear the hot tier")
+def step_when_clear(context: Any) -> None:
+    context.hot_tier.clear()
+
+
+# ---------------------------------------------------------------------------
+# Metrics assertions
+# ---------------------------------------------------------------------------
+
+
+@then("the hot tier entry_count should be {n:d}")
+def step_then_entry_count(context: Any, n: int) -> None:
+    assert context.hot_tier.entry_count == n, (
+        f"Expected entry_count={n}, got {context.hot_tier.entry_count}"
+    )
+
+
+@then("the hot storage tier size_bytes should be {n:d}")
+def step_then_size_bytes(context: Any, n: int) -> None:
+    assert context.hot_tier.size_bytes == n, (
+        f"Expected size_bytes={n}, got {context.hot_tier.size_bytes}"
+    )
+
+
+@then("the hot storage tier size_bytes should be at most {n:d}")
+def step_then_size_bytes_at_most(context: Any, n: int) -> None:
+    assert context.hot_tier.size_bytes <= n, (
+        f"Expected size_bytes <= {n}, got {context.hot_tier.size_bytes}"
+    )
+
+
+@then("the hot tier hit_count should be {n:d}")
+def step_then_hit_count(context: Any, n: int) -> None:
+    assert context.hot_tier.hit_count == n, (
+        f"Expected hit_count={n}, got {context.hot_tier.hit_count}"
+    )
+
+
+@then("the hot tier miss_count should be {n:d}")
+def step_then_miss_count(context: Any, n: int) -> None:
+    assert context.hot_tier.miss_count == n, (
+        f"Expected miss_count={n}, got {context.hot_tier.miss_count}"
+    )
+
+
+@then("the hot tier max_entries should be {n:d}")
+def step_then_max_entries(context: Any, n: int) -> None:
+    assert context.hot_tier.max_entries == n, (
+        f"Expected max_entries={n}, got {context.hot_tier.max_entries}"
+    )
+
+
+@then("the hot tier max_entries should be None")
+def step_then_max_entries_none(context: Any) -> None:
+    assert context.hot_tier.max_entries is None, (
+        f"Expected max_entries=None, got {context.hot_tier.max_entries}"
+    )
+
+
+@then("the hot tier max_bytes should be {n:d}")
+def step_then_max_bytes(context: Any, n: int) -> None:
+    assert context.hot_tier.max_bytes == n, (
+        f"Expected max_bytes={n}, got {context.hot_tier.max_bytes}"
+    )
+
+
+@then("the hot tier max_bytes should be None")
+def step_then_max_bytes_none(context: Any) -> None:
+    assert context.hot_tier.max_bytes is None, (
+        f"Expected max_bytes=None, got {context.hot_tier.max_bytes}"
+    )
+
+
+@then("the hot tier entry_count should be at most {n:d}")
+def step_then_entry_count_at_most(context: Any, n: int) -> None:
+    assert context.hot_tier.entry_count <= n, (
+        f"Expected entry_count <= {n}, got {context.hot_tier.entry_count}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Eviction callback assertions
+# ---------------------------------------------------------------------------
+
+
+@then('the eviction recorder should have recorded entry id "{entry_id}"')
+def step_then_eviction_recorded_id(context: Any, entry_id: str) -> None:
+    recorded_ids = [e[0] for e in context.evicted_entries]
+    assert entry_id in recorded_ids, (
+        f"Expected {entry_id!r} in evicted entries, got {recorded_ids}"
+    )
+
+
+@then(
+    'the eviction recorder should have recorded entry "{entry_id}" '
+    'with content "{content}"'
+)
+def step_then_eviction_recorded_with_content(
+    context: Any, entry_id: str, content: str
+) -> None:
+    for eid, econtent in context.evicted_entries:
+        if eid == entry_id:
+            assert econtent == content, (
+                f"Expected content {content!r} for evicted {entry_id!r}, "
+                f"got {econtent!r}"
+            )
+            return
+    raise AssertionError(
+        f"Entry {entry_id!r} not found in evicted entries: {context.evicted_entries}"
+    )
+
+
+@then("the eviction recorder should have recorded {n:d} evictions")
+def step_then_eviction_count(context: Any, n: int) -> None:
+    assert len(context.evicted_entries) == n, (
+        f"Expected {n} evictions, got {len(context.evicted_entries)}: "
+        f"{context.evicted_entries}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Thread safety
+# ---------------------------------------------------------------------------
+
+
+@when(
+    "{num_threads:d} threads concurrently put {entries_per_thread:d} entries each "
+    "into the hot tier"
+)
+def step_when_concurrent_puts(
+    context: Any, num_threads: int, entries_per_thread: int
+) -> None:
+    context.concurrent_exception: Exception | None = None
+    barrier = threading.Barrier(num_threads)
+
+    def worker(thread_idx: int) -> None:
+        try:
+            barrier.wait()
+            for i in range(entries_per_thread):
+                key = f"thread-{thread_idx}-entry-{i}"
+                context.hot_tier.put(key, f"content-{thread_idx}-{i}")
+        except Exception as exc:
+            context.concurrent_exception = exc
+
+    threads = [threading.Thread(target=worker, args=(i,)) for i in range(num_threads)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+
+@when(
+    "{num_threads:d} threads concurrently put entries and "
+    "{num_threads2:d} threads concurrently get entries"
+)
+def step_when_concurrent_puts_and_gets(
+    context: Any, num_threads: int, num_threads2: int
+) -> None:
+    context.concurrent_exception = None
+    total_threads = num_threads + num_threads2
+    barrier = threading.Barrier(total_threads)
+
+    # Pre-populate some entries for getters to find
+    for i in range(20):
+        context.hot_tier.put(f"pre-{i}", f"pre-content-{i}")
+
+    def putter(thread_idx: int) -> None:
+        try:
+            barrier.wait()
+            for i in range(10):
+                context.hot_tier.put(f"put-{thread_idx}-{i}", f"val-{i}")
+        except Exception as exc:
+            context.concurrent_exception = exc
+
+    def getter(thread_idx: int) -> None:
+        try:
+            barrier.wait()
+            for i in range(10):
+                context.hot_tier.get(f"pre-{i % 20}")
+        except Exception as exc:
+            context.concurrent_exception = exc
+
+    threads = [
+        threading.Thread(target=putter, args=(i,)) for i in range(num_threads)
+    ] + [threading.Thread(target=getter, args=(i,)) for i in range(num_threads2)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+
+@when("{num_threads:d} threads concurrently put large entries into the hot tier")
+def step_when_concurrent_large_puts(context: Any, num_threads: int) -> None:
+    context.concurrent_exception = None
+    barrier = threading.Barrier(num_threads)
+
+    def worker(thread_idx: int) -> None:
+        try:
+            barrier.wait()
+            for i in range(5):
+                key = f"large-{thread_idx}-{i}"
+                # Each entry is ~20 bytes, well above max_bytes=100 per thread
+                context.hot_tier.put(key, f"content-{thread_idx:03d}-{i:03d}-pad")
+        except Exception as exc:
+            context.concurrent_exception = exc
+
+    threads = [threading.Thread(target=worker, args=(i,)) for i in range(num_threads)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+
+@then("no exception should have been raised during concurrent puts")
+def step_then_no_concurrent_put_exception(context: Any) -> None:
+    assert context.concurrent_exception is None, (
+        f"Concurrent puts raised: {context.concurrent_exception}"
+    )
+
+
+@then("no exception should have been raised during concurrent gets and puts")
+def step_then_no_concurrent_get_put_exception(context: Any) -> None:
+    assert context.concurrent_exception is None, (
+        f"Concurrent gets/puts raised: {context.concurrent_exception}"
+    )
+
+
+@then("no exception should have been raised during concurrent evictions")
+def step_then_no_concurrent_eviction_exception(context: Any) -> None:
+    assert context.concurrent_exception is None, (
+        f"Concurrent evictions raised: {context.concurrent_exception}"
+    )

src/cleveragents/acms/__init__.py

@@ -6,7 +6,8 @@ inheritance mechanism for resolving named detail levels across the
 ontology hierarchy (Layer 3 -> Layer 2 -> Layer 1 -> Layer 0).
 
 Also provides the ACMS index data model and file traversal engine for
-indexing large projects.
+indexing large projects, and the hot storage tier LRU cache
+implementation.
 
 Based on ``docs/specification.md`` ~lines 42333-42422, 44405-44420.
 """
@@ -21,6 +22,7 @@ from cleveragents.acms.index import (
     IndexEntry,
     TierLevel,
 )
+from cleveragents.acms.storage.hot import HotStorageTier
 from cleveragents.acms.uko import (
     CODE_DETAIL_LEVEL_MAP,
     FUNC_DETAIL_LEVEL_MAP,
@@ -72,7 +74,7 @@ from cleveragents.acms.uko import (
     resolve_detail_level,
 )
 
-# Combine exports from both uko and index modules
+# Combine exports from uko, index, and storage modules
 _uko_exports = list(_uko.__all__)
 _index_exports = [
     "ACMSIndex",
@@ -81,5 +83,6 @@ _index_exports = [
     "IndexEntry",
     "TierLevel",
 ]
+_storage_exports = ["HotStorageTier"]
 
-__all__: list[str] = _uko_exports + _index_exports
+__all__: list[str] = _uko_exports + _index_exports + _storage_exports

src/cleveragents/acms/storage/__init__.py

@@ -0,0 +1,17 @@
+"""ACMS storage tier implementations.
+
+Provides concrete storage tier classes for the Advanced Context
+Management System (ACMS):
+
+| Class             | Role                                                    |
+|-------------------|---------------------------------------------------------|
+| ``HotStorageTier``| In-memory LRU cache with configurable capacity limits   |
+
+Based on ``docs/specification.md`` ACMS tier sections and issue #9972.
+"""
+
+from __future__ import annotations
+
+from cleveragents.acms.storage.hot import HotStorageTier
+
+__all__: list[str] = ["HotStorageTier"]

src/cleveragents/acms/storage/hot.py

@@ -0,0 +1,260 @@
+"""Hot storage tier: in-memory LRU cache with configurable capacity.
+
+The hot storage tier is the primary access layer for the ACMS, holding
+recently and frequently accessed context entries in memory for fast
+retrieval.  It is backed by an ``OrderedDict`` to maintain LRU ordering
+with O(1) get/put operations.
+
+Capacity is controlled by two independent limits:
+
+* ``max_entries`` — maximum number of entries in the cache.
+* ``max_bytes``   — maximum total byte size of all cached content.
+
+When either limit is exceeded, the least-recently-used entry is evicted
+first.  An optional ``on_evict`` callback is invoked for each evicted
+entry, enabling warm-tier demotion.
+
+Thread safety is provided by a ``threading.RLock`` so that concurrent
+reads and writes from parallel plan execution do not corrupt the cache.
+
+Based on ``docs/specification.md`` ACMS tier sections and issue #9972.
+"""
+
+from __future__ import annotations
+
+import threading
+from collections import OrderedDict
+from collections.abc import Callable
+
+import structlog
+
+logger = structlog.get_logger(__name__)
+
+
+class HotStorageTier:
+    """In-memory LRU cache for ACMS hot-tier context entries.
+
+    Entries are keyed by a string ``entry_id`` and store arbitrary
+    ``str`` content.  The cache tracks byte sizes using UTF-8 encoding
+    of the content string.
+
+    Args:
+        max_entries: Maximum number of entries.  ``None`` means no limit.
+        max_bytes:   Maximum total byte size of all content.  ``None``
+                     means no limit.
+        on_evict:    Optional callback invoked with ``(entry_id, content)``
+                     when an entry is evicted due to capacity overflow.
+                     Intended for warm-tier demotion.  Errors raised by
+                     the callback are logged and suppressed so that cache
+                     operations are never interrupted by callback failures.
+
+    Raises:
+        ValueError: If ``max_entries`` is not ``None`` and is less than 1.
+        ValueError: If ``max_bytes`` is not ``None`` and is less than 1.
+    """
+
+    def __init__(
+        self,
+        max_entries: int | None = None,
+        max_bytes: int | None = None,
+        on_evict: Callable[[str, str], None] | None = None,
+    ) -> None:
+        if max_entries is not None and max_entries < 1:
+            raise ValueError(f"max_entries must be at least 1, got {max_entries}")
+        if max_bytes is not None and max_bytes < 1:
+            raise ValueError(f"max_bytes must be at least 1, got {max_bytes}")
+
+        self._max_entries = max_entries
+        self._max_bytes = max_bytes
+        self._on_evict = on_evict
+
+        # OrderedDict preserves insertion order; we move accessed keys to
+        # the end (most-recently-used) so the front is always the LRU entry.
+        self._cache: OrderedDict[str, str] = OrderedDict()
+        self._size_bytes: int = 0
+
+        self._hit_count: int = 0
+        self._miss_count: int = 0
+
+        self._lock = threading.RLock()
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    def put(self, entry_id: str, content: str) -> None:
+        """Insert or update an entry in the cache.
+
+        If the entry already exists it is updated in-place and moved to
+        the most-recently-used position.  After insertion, capacity
+        limits are enforced by evicting LRU entries until both limits
+        are satisfied.
+
+        Args:
+            entry_id: Unique identifier for the entry.
+            content:  String content to cache.
+
+        Raises:
+            ValueError: If ``entry_id`` is empty.
+        """
+        if not entry_id:
+            raise ValueError("entry_id must be non-empty")
+
+        content_bytes = len(content.encode("utf-8"))
+
+        with self._lock:
+            # If the entry already exists, remove it first so we can
+            # re-insert at the MRU end and update the byte accounting.
+            if entry_id in self._cache:
+                old_content = self._cache[entry_id]
+                self._size_bytes -= len(old_content.encode("utf-8"))
+                del self._cache[entry_id]
+
+            self._cache[entry_id] = content
+            self._size_bytes += content_bytes
+
+            # Move to MRU end (already there since we just inserted, but
+            # explicit for clarity).
+            self._cache.move_to_end(entry_id)
+
+            self._enforce_capacity()
+
+    def get(self, entry_id: str) -> str | None:
+        """Retrieve an entry, updating its LRU position.
+
+        Returns the cached content string, or ``None`` if the entry is
+        not present.  Updates hit/miss counters.
+
+        Args:
+            entry_id: Unique identifier for the entry.
+        """
+        with self._lock:
+            if entry_id in self._cache:
+                self._hit_count += 1
+                # Move to MRU end
+                self._cache.move_to_end(entry_id)
+                return self._cache[entry_id]
+
+            self._miss_count += 1
+            return None
+
+    def remove(self, entry_id: str) -> str | None:
+        """Remove an entry from the cache without invoking the evict callback.
+
+        Returns the removed content, or ``None`` if the entry was not
+        present.
+
+        Args:
+            entry_id: Unique identifier for the entry.
+        """
+        with self._lock:
+            if entry_id not in self._cache:
+                return None
+            content = self._cache.pop(entry_id)
+            self._size_bytes -= len(content.encode("utf-8"))
+            return content
+
+    def clear(self) -> None:
+        """Remove all entries from the cache.
+
+        Does **not** invoke the ``on_evict`` callback for cleared entries.
+        Resets ``size_bytes`` to zero but preserves hit/miss counters.
+        """
+        with self._lock:
+            self._cache.clear()
+            self._size_bytes = 0
+
+    # ------------------------------------------------------------------
+    # Metrics (read-only properties)
+    # ------------------------------------------------------------------
+
+    @property
+    def hit_count(self) -> int:
+        """Total number of cache hits since construction."""
+        with self._lock:
+            return self._hit_count
+
+    @property
+    def miss_count(self) -> int:
+        """Total number of cache misses since construction."""
+        with self._lock:
+            return self._miss_count
+
+    @property
+    def entry_count(self) -> int:
+        """Current number of entries in the cache."""
+        with self._lock:
+            return len(self._cache)
+
+    @property
+    def size_bytes(self) -> int:
+        """Current total byte size of all cached content (UTF-8 encoded)."""
+        with self._lock:
+            return self._size_bytes
+
+    # ------------------------------------------------------------------
+    # Configuration accessors
+    # ------------------------------------------------------------------
+
+    @property
+    def max_entries(self) -> int | None:
+        """Configured maximum entry count limit (``None`` = unlimited)."""
+        return self._max_entries
+
+    @property
+    def max_bytes(self) -> int | None:
+        """Configured maximum byte size limit (``None`` = unlimited)."""
+        return self._max_bytes
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+
+    def _enforce_capacity(self) -> None:
+        """Evict LRU entries until both capacity limits are satisfied.
+
+        Called after every ``put`` operation.  Evicts from the front of
+        the ``OrderedDict`` (least-recently-used end).
+        """
+        while self._cache and self._is_over_capacity():
+            self._evict_lru()
+
+    def _is_over_capacity(self) -> bool:
+        """Return ``True`` if either capacity limit is currently exceeded."""
+        if self._max_entries is not None and len(self._cache) > self._max_entries:
+            return True
+        return self._max_bytes is not None and self._size_bytes > self._max_bytes
+
+    def _evict_lru(self) -> None:
+        """Evict the single least-recently-used entry.
+
+        Invokes ``on_evict`` callback if configured.  Callback errors
+        are caught, logged, and suppressed.
+        """
+        if not self._cache:
+            return
+
+        # The LRU entry is at the front of the OrderedDict.
+        lru_id, lru_content = next(iter(self._cache.items()))
+        del self._cache[lru_id]
+        self._size_bytes -= len(lru_content.encode("utf-8"))
+
+        logger.debug(
+            "hot_storage.evicted",
+            entry_id=lru_id,
+            remaining_entries=len(self._cache),
+            remaining_bytes=self._size_bytes,
+        )
+
+        if self._on_evict is not None:
+            try:
+                self._on_evict(lru_id, lru_content)
+            except Exception:
+                logger.warning(
+                    "hot_storage.evict_callback_failed",
+                    entry_id=lru_id,
+                    exc_info=True,
+                )
+
+
+__all__: list[str] = ["HotStorageTier"]