feat(context): add hot/warm/cold tiers and actor views

Implement hot/warm/cold context tiers with ContextTier, ActorRole,
TieredFragment, TierBudget, ActorContextView, TierMetrics, and
ScopedBackendView models. ContextTierService provides store/get,
promotion/demotion with cold-tier summarisation hook, LRU eviction,
per-actor filtered views (strategist/executor/reviewer), and
project-scoped isolation via ScopedBackendView.

Settings: context_max_tokens_hot, context_max_decisions_warm,
context_max_decisions_cold. DI-wired as singleton context_tier_service.

Includes Behave BDD scenarios (30), Robot Framework integration tests (7),
ASV benchmarks (5 suites), and docs/reference/context_tiers.md.

Closes #208

CHANGELOG.md

@@ -2,6 +2,14 @@
 
 ## Unreleased
 
+- Added hot/warm/cold context tiers with `ContextTier`, `ActorRole`, `TieredFragment`,
+  `TierBudget`, `ActorContextView`, `TierMetrics`, and `ScopedBackendView` models.
+  `ContextTierService` provides store/get, promotion/demotion with cold-tier summarisation
+  hook, LRU eviction, per-actor filtered views (strategist/executor/reviewer), and
+  project-scoped isolation. Settings: `context_max_tokens_hot`, `context_max_decisions_warm`,
+  `context_max_decisions_cold`. DI-wired as singleton `context_tier_service`. Includes Behave
+  BDD scenarios, Robot Framework integration tests, ASV benchmarks, and
+  `docs/reference/context_tiers.md`. (#208)
 - Added `SafetyProfile` domain model with configurable safety constraints (allowed skill
   categories, sandbox/checkpoint requirements, human-approval flag, cost/retry limits) and
   integrated it into the `Action` model via `from_config`/`as_cli_dict`. Persistence backed

benchmarks/bench_coverage_report.py

@@ -202,5 +202,5 @@ class CoverageFileSuite:
         return 0
 
 
-setattr(CoverageFileSuite.track_source_file_count, "unit", "files")
-setattr(CoverageFileSuite.track_source_total_bytes, "unit", "bytes")
+CoverageFileSuite.track_source_file_count.unit = "files"
+CoverageFileSuite.track_source_total_bytes.unit = "bytes"

benchmarks/bench_subprocess_overhead.py

@@ -52,9 +52,9 @@ class SubprocessCountSuite:
         return 0
 
 
-setattr(SubprocessCountSuite.track_subprocess_count, "unit", "subprocesses")
-setattr(SubprocessCountSuite.track_recursive_feature_count, "unit", "features")
-setattr(SubprocessCountSuite.track_step_definition_count, "unit", "files")
+SubprocessCountSuite.track_subprocess_count.unit = "subprocesses"
+SubprocessCountSuite.track_recursive_feature_count.unit = "features"
+SubprocessCountSuite.track_step_definition_count.unit = "files"
 
 
 class PerFeatureOverheadSuite:
@@ -149,5 +149,5 @@ class WorkerPoolSizingSuite:
         return max(1, math.ceil(self.feature_count / cpus))
 
 
-setattr(WorkerPoolSizingSuite.track_features_per_cpu, "unit", "features/cpu")
-setattr(WorkerPoolSizingSuite.track_optimal_chunk_size, "unit", "features/chunk")
+WorkerPoolSizingSuite.track_features_per_cpu.unit = "features/cpu"
+WorkerPoolSizingSuite.track_optimal_chunk_size.unit = "features/chunk"

benchmarks/bench_unit_tests.py

@@ -87,8 +87,8 @@ class StepModuleDiscoverySuite:
         return sum(f.stat().st_size for f in self.steps_dir.glob("*.py"))
 
 
-setattr(StepModuleDiscoverySuite.track_step_file_count, "unit", "files")
-setattr(StepModuleDiscoverySuite.track_step_total_bytes, "unit", "bytes")
+StepModuleDiscoverySuite.track_step_file_count.unit = "files"
+StepModuleDiscoverySuite.track_step_total_bytes.unit = "bytes"
 
 
 class ParallelChunkSuite:
@@ -195,6 +195,6 @@ class TrackTestSuiteMetrics:
         return count
 
 
-setattr(TrackTestSuiteMetrics.track_feature_file_count, "unit", "features")
-setattr(TrackTestSuiteMetrics.track_total_feature_bytes, "unit", "bytes")
-setattr(TrackTestSuiteMetrics.track_scenario_line_count, "unit", "scenarios")
+TrackTestSuiteMetrics.track_feature_file_count.unit = "features"
+TrackTestSuiteMetrics.track_total_feature_bytes.unit = "bytes"
+TrackTestSuiteMetrics.track_scenario_line_count.unit = "scenarios"

benchmarks/context_tiers_bench.py

@@ -0,0 +1,168 @@
+"""ASV benchmarks for ACMS Context Tiers.
+
+Measures the performance of:
+- TieredFragment construction
+- ContextTierService store/get operations
+- Tier promotion and demotion
+- LRU eviction
+- Actor view filtering
+- Project scoping
+"""
+
+from __future__ import annotations
+
+import importlib
+import sys
+from pathlib import Path
+
+# Ensure the local *source* tree is importable even when ASV has an
+# older build of the package installed.
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+# Force-reload so ASV picks up the source tree version.
+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
+    ActorRole,
+    ContextTier,
+    ScopedBackendView,
+    TierBudget,
+    TieredFragment,
+    TierMetrics,
+)
+
+# ---------------------------------------------------------------------------
+# Model construction benchmarks
+# ---------------------------------------------------------------------------
+
+
+class TierModelConstructionSuite:
+    """Benchmark tier model object creation overhead."""
+
+    timeout = 60
+
+    def time_create_tiered_fragment(self) -> None:
+        TieredFragment(fragment_id="bench-001", content="benchmark data")
+
+    def time_create_tier_budget(self) -> None:
+        TierBudget()
+
+    def time_create_tier_metrics(self) -> None:
+        TierMetrics()
+
+    def time_create_scoped_view(self) -> None:
+        ScopedBackendView(allowed_projects=["proj-a", "proj-b"])
+
+
+# ---------------------------------------------------------------------------
+# Service operation benchmarks
+# ---------------------------------------------------------------------------
+
+
+class TierServiceStoreSuite:
+    """Benchmark ContextTierService store/get operations."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.service = ContextTierService()
+        self.fragment = TieredFragment(
+            fragment_id="bench-store",
+            content="benchmark content",
+            tier=ContextTier.HOT,
+        )
+
+    def time_store_fragment(self) -> None:
+        self.service.store(self.fragment)
+
+    def time_get_fragment(self) -> None:
+        self.service.store(self.fragment)
+        self.service.get("bench-store")
+
+    def time_get_miss(self) -> None:
+        self.service.get("nonexistent")
+
+
+class TierServicePromotionSuite:
+    """Benchmark tier promotion and demotion."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.service = ContextTierService()
+
+    def time_promote_cold_to_warm(self) -> None:
+        frag = TieredFragment(
+            fragment_id="promote-bench",
+            content="data",
+            tier=ContextTier.COLD,
+        )
+        self.service.store(frag)
+        self.service.promote("promote-bench")
+
+    def time_demote_hot_to_warm(self) -> None:
+        frag = TieredFragment(
+            fragment_id="demote-bench",
+            content="data",
+            tier=ContextTier.HOT,
+        )
+        self.service.store(frag)
+        self.service.demote("demote-bench")
+
+
+class TierServiceEvictionSuite:
+    """Benchmark LRU eviction performance."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.service = ContextTierService()
+        for i in range(100):
+            self.service.store(
+                TieredFragment(
+                    fragment_id=f"evict-{i}",
+                    content=f"data-{i}",
+                    tier=ContextTier.HOT,
+                )
+            )
+
+    def time_evict_10_from_100(self) -> None:
+        self.service.evict_lru(ContextTier.HOT, 10)
+
+
+class TierServiceActorViewSuite:
+    """Benchmark per-actor filtered view performance."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.service = ContextTierService()
+        for i in range(50):
+            tier = [ContextTier.HOT, ContextTier.WARM, ContextTier.COLD][i % 3]
+            self.service.store(
+                TieredFragment(
+                    fragment_id=f"actor-{i}",
+                    content=f"data-{i}",
+                    tier=tier,
+                    project_name="bench-proj",
+                )
+            )
+
+    def time_strategist_view(self) -> None:
+        self.service.get_for_actor(ActorRole.STRATEGIST, ["bench-proj"])
+
+    def time_executor_view(self) -> None:
+        self.service.get_for_actor(ActorRole.EXECUTOR, ["bench-proj"])
+
+    def time_reviewer_view(self) -> None:
+        self.service.get_for_actor(ActorRole.REVIEWER, ["bench-proj"])
+
+    def time_scoped_view(self) -> None:
+        self.service.get_scoped_view(["bench-proj"])

benchmarks/decision_di_bench.py

@@ -32,7 +32,7 @@ except ModuleNotFoundError:
     from cleveragents.application.services.decision_service import DecisionService
     from cleveragents.config.settings import Settings
     from cleveragents.domain.models.core.action import Action, ActionState
-    from cleveragents.domain.models.core.decision import Decision, DecisionType
+    from cleveragents.domain.models.core.decision import DecisionType
     from cleveragents.domain.models.core.plan import (
         NamespacedName,
         Plan,

benchmarks/devcontainer_handler_bench.py

@@ -29,9 +29,6 @@ try:
     )
 except ModuleNotFoundError:
     sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
-    from cleveragents.resource.handlers.devcontainer import (
-        DevcontainerHandler,
-    )
     from cleveragents.resource.handlers.discovery import (
         DevcontainerDiscoveryResult,
         discover_devcontainers,

benchmarks/permission_check_bench.py

@@ -14,7 +14,6 @@ from cleveragents.application.services.permission_service import (
 from cleveragents.domain.models.core.permission import (
     ROLE_PERMISSIONS,
     PermissionAction,
-    PermissionPolicy,
     PermissionRole,
     PermissionScope,
     RoleBinding,

benchmarks/semantic_validation_bench.py

@@ -11,7 +11,6 @@ from __future__ import annotations
 
 import sys
 from pathlib import Path
-from typing import Any
 
 _SRC = str(Path(__file__).resolve().parents[1] / "src")
 if _SRC not in sys.path:
@@ -28,7 +27,6 @@ from cleveragents.application.services.semantic_validation_rules import (
     SyntaxCheckRule,
 )
 from cleveragents.application.services.semantic_validation_service import (
-    SemanticRuleRegistry,
     SemanticValidationCache,
     SemanticValidationService,
     create_default_registry,

benchmarks/server_stub_bench.py

@@ -31,7 +31,6 @@ from cleveragents.acp.clients import (  # noqa: E402
 from cleveragents.acp.server_config import ServerConnectionConfig  # noqa: E402
 from cleveragents.application.services.config_service import _REGISTRY  # noqa: E402
 
-
 # ---------------------------------------------------------------------------
 # Config resolution benchmarks
 # ---------------------------------------------------------------------------

benchmarks/subplan_execution_bench.py

@@ -16,7 +16,6 @@ from cleveragents.application.services.subplan_merge_service import (
 )
 from cleveragents.domain.models.core.plan import (
     ExecutionMode,
-    ProcessingState,
     SubplanConfig,
     SubplanMergeStrategy,
     SubplanStatus,

docs/reference/context_tiers.md

@@ -0,0 +1,109 @@
+# Context Tiers
+
+The ACMS Context Tier system manages context fragments across three storage
+tiers — **hot**, **warm**, and **cold** — with per-actor visibility, project
+scoping, and LRU eviction.
+
+## Overview
+
+| Tier   | Backend     | Latency | Capacity         |
+|--------|-------------|---------|------------------|
+| Hot    | In-memory   | Low     | Token-budget     |
+| Warm   | SQLite stub | Medium  | Decision-count   |
+| Cold   | File stub   | High    | Decision-count   |
+
+## Models
+
+### ContextTier
+
+```python
+class ContextTier(StrEnum):
+    HOT = "hot"
+    WARM = "warm"
+    COLD = "cold"
+```
+
+### ActorRole
+
+```python
+class ActorRole(StrEnum):
+    STRATEGIST = "strategist"  # sees hot + warm + cold
+    EXECUTOR = "executor"      # sees hot + warm
+    REVIEWER = "reviewer"      # sees hot only
+```
+
+### TieredFragment
+
+Extends the CRP `ContextFragment` concept with tier placement and access
+tracking metadata:
+
+- `fragment_id` — stable unique identifier
+- `content` — rendered text content
+- `tier` — current storage tier
+- `resource_id` — originating resource
+- `project_name` — project scope for isolation
+- `token_count` — token count of content
+- `last_accessed` — timestamp for LRU eviction
+- `access_count` — access frequency counter
+- `created_at` — creation timestamp
+
+### TierBudget
+
+Controls capacity per tier:
+
+- `max_tokens_hot` (default: 8000)
+- `max_decisions_warm` (default: 500)
+- `max_decisions_cold` (default: 5000)
+
+### ActorContextView
+
+Per-actor filtered view configuration with role-based default tiers.
+
+### TierMetrics
+
+Hit/miss counters: `hot_hit_count`, `hot_miss_count`, `warm_hit_count`,
+`warm_miss_count`, plus population counts per tier.
+
+### ScopedBackendView
+
+Project-scoped filter that enforces resource isolation. Fragments with
+empty `project_name` are excluded by default.
+
+## Service
+
+### ContextTierService
+
+```python
+from cleveragents.application.services.context_tiers import ContextTierService
+
+svc = ContextTierService()
+svc.store(fragment)
+svc.get("fragment-id")
+svc.promote("fragment-id")     # cold→warm or warm→hot
+svc.demote("fragment-id")      # hot→warm or warm→cold (with summarisation)
+svc.evict_lru(ContextTier.HOT, count=5)
+svc.get_for_actor(ActorRole.STRATEGIST, project_names=["my-project"])
+svc.get_scoped_view(["my-project"])
+svc.get_metrics()
+```
+
+## Configuration
+
+Environment variables (via `Settings`):
+
+| Variable | Default | Description |
+|----------|---------|-------------|
+| `CLEVERAGENTS_CONTEXT_MAX_TOKENS_HOT` | 8000 | Max tokens in hot tier |
+| `CLEVERAGENTS_CONTEXT_MAX_DECISIONS_WARM` | 500 | Max fragments in warm tier |
+| `CLEVERAGENTS_CONTEXT_MAX_DECISIONS_COLD` | 5000 | Max fragments in cold tier |
+
+## DI Container
+
+The `ContextTierService` is registered as a singleton in the DI container:
+
+```python
+from cleveragents.application.container import get_container
+
+container = get_container()
+svc = container.context_tier_service()
+```

features/context_tiers.feature

@@ -0,0 +1,191 @@
+Feature: ACMS Context Tiers
+  As a developer
+  I want hot/warm/cold context tiers with actor views and project scoping
+  So that the context management system can efficiently manage fragment lifecycle
+
+  # ---- ContextTier enum ----
+
+  Scenario: ContextTier has three values
+    Then ContextTier should have values "hot", "warm", "cold"
+
+  # ---- ActorRole enum ----
+
+  Scenario: ActorRole has three values
+    Then ActorRole should have values "strategist", "executor", "reviewer"
+
+  # ---- TieredFragment ----
+
+  Scenario: Create a valid TieredFragment
+    Given a TieredFragment with id "frag-001" and content "hello world"
+    Then the fragment tier should be "hot"
+    And the fragment token_count should be 0
+    And the fragment access_count should be 0
+
+  Scenario: TieredFragment rejects empty fragment_id
+    Then creating a TieredFragment with empty fragment_id should raise ValidationError
+
+  Scenario: TieredFragment rejects negative token_count
+    Then creating a TieredFragment with negative token_count should raise ValidationError
+
+  # ---- TierBudget ----
+
+  Scenario: TierBudget has sensible defaults
+    Given a default TierBudget
+    Then max_tokens_hot should be 8000
+    And max_decisions_warm should be 500
+    And max_decisions_cold should be 5000
+
+  # ---- ActorContextView ----
+
+  Scenario: Strategist sees all tiers by default
+    Given an ActorContextView for "strategist"
+    Then effective tiers should be "hot", "warm", "cold"
+
+  Scenario: Executor sees hot and warm by default
+    Given an ActorContextView for "executor"
+    Then effective tiers should be "hot", "warm"
+
+  Scenario: Reviewer sees only hot by default
+    Given an ActorContextView for "reviewer"
+    Then effective tiers should be "hot"
+
+  # ---- TierMetrics ----
+
+  Scenario: TierMetrics defaults to zero
+    Given a default TierMetrics
+    Then total_fragments should be 0
+    And hot_hit_rate should be 0.0
+
+  # ---- ScopedBackendView ----
+
+  Scenario: ScopedBackendView filters by project
+    Given a ScopedBackendView for project "proj-a"
+    And a TieredFragment with id "frag-a" in project "proj-a"
+    And a TieredFragment with id "frag-b" in project "proj-b"
+    Then fragment "frag-a" should be visible
+    And fragment "frag-b" should not be visible
+
+  Scenario: ScopedBackendView excludes empty project_name
+    Given a ScopedBackendView for project "proj-a"
+    And a TieredFragment with id "frag-x" and no project
+    Then fragment "frag-x" should not be visible
+
+  # ---- ContextTierService: store/get ----
+
+  Scenario: Store and retrieve a fragment
+    Given a ContextTierService
+    When I store a fragment with id "f1" content "data" in tier "hot"
+    Then tier getting "f1" should return content "data"
+
+  Scenario: Get returns None for unknown fragment
+    Given a ContextTierService
+    Then tier getting "nonexistent" should return None
+
+  # ---- ContextTierService: promotion ----
+
+  Scenario: Promote from cold to warm
+    Given a ContextTierService
+    When I store a fragment with id "f2" content "cold-data" in tier "cold"
+    And I promote "f2"
+    Then "f2" should be in tier "warm"
+
+  Scenario: Promote from warm to hot
+    Given a ContextTierService
+    When I store a fragment with id "f3" content "warm-data" in tier "warm"
+    And I promote "f3"
+    Then "f3" should be in tier "hot"
+
+  Scenario: Promote unknown fragment returns None
+    Given a ContextTierService
+    Then promoting "unknown" should return None
+
+  # ---- ContextTierService: demotion ----
+
+  Scenario: Demote from hot to warm
+    Given a ContextTierService
+    When I store a fragment with id "f4" content "hot-data" in tier "hot"
+    And I demote "f4"
+    Then "f4" should be in tier "warm"
+
+  Scenario: Demote from warm to cold with summarization
+    Given a ContextTierService
+    When I store a long fragment "f5" in tier "warm"
+    And I demote "f5"
+    Then "f5" should be in tier "cold"
+    And "f5" content should be summarized
+
+  Scenario: Demote unknown fragment returns None
+    Given a ContextTierService
+    Then demoting "unknown" should return None
+
+  # ---- ContextTierService: LRU eviction ----
+
+  Scenario: Evict LRU fragments from hot tier
+    Given a ContextTierService
+    When I store a fragment with id "e1" content "first" in tier "hot"
+    And I store a fragment with id "e2" content "second" in tier "hot"
+    And I store a fragment with id "e3" content "third" in tier "hot"
+    And I evict 2 LRU fragments from "hot"
+    Then "hot" tier should have 1 fragment
+
+  Scenario: Evict with invalid count raises ValueError
+    Given a ContextTierService
+    Then evicting 0 from "hot" should raise ValueError
+
+  # ---- ContextTierService: actor views ----
+
+  Scenario: Strategist sees fragments from all tiers
+    Given a ContextTierService
+    When I store fragment "a1" with content "hot" tier "hot" project "p1"
+    And I store fragment "a2" with content "warm" tier "warm" project "p1"
+    And I store fragment "a3" with content "cold" tier "cold" project "p1"
+    Then strategist for project "p1" should see 3 fragments
+
+  Scenario: Executor sees only hot and warm
+    Given a ContextTierService
+    When I store fragment "b1" with content "hot" tier "hot" project "p1"
+    And I store fragment "b2" with content "warm" tier "warm" project "p1"
+    And I store fragment "b3" with content "cold" tier "cold" project "p1"
+    Then executor for project "p1" should see 2 fragments
+
+  Scenario: Reviewer sees only hot
+    Given a ContextTierService
+    When I store fragment "c1" with content "hot" tier "hot" project "p1"
+    And I store fragment "c2" with content "warm" tier "warm" project "p1"
+    Then reviewer for project "p1" should see 1 fragment
+
+  # ---- ContextTierService: project scoping ----
+
+  Scenario: Scoped view filters by project
+    Given a ContextTierService
+    When I store fragment "s1" with content "yes" tier "hot" project "alpha"
+    And I store fragment "s2" with content "no" tier "hot" project "beta"
+    Then scoped view for "alpha" should have 1 fragment
+
+  # ---- ContextTierService: metrics ----
+
+  Scenario: Metrics track hits and misses
+    Given a ContextTierService
+    When I store a fragment with id "m1" content "data" in tier "hot"
+    And I get fragment "m1"
+    And I get fragment "nonexistent"
+    Then hot_hit_count should be 1
+    And hot_miss_count should be 1
+
+  # ---- DI Container ----
+
+  Scenario: DI container provides context_tier_service
+    Given the DI container is reset
+    Then the container should provide a context_tier_service
+    And the context_tier_service should be a ContextTierService
+
+  Scenario: DI container context_tier_service is singleton
+    Given the DI container is reset
+    Then resolving context_tier_service twice should return the same instance
+
+  # ---- Settings ----
+
+  Scenario: Settings include context tier budget fields
+    Then settings should have context_max_tokens_hot
+    And settings should have context_max_decisions_warm
+    And settings should have context_max_decisions_cold

features/steps/context_tiers_steps.py

@@ -0,0 +1,450 @@
+"""Step definitions for the ACMS Context Tiers feature."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from behave import given, then, when
+from pydantic import ValidationError
+
+from cleveragents.application.container import get_container, reset_container
+from cleveragents.application.services.context_tiers import ContextTierService
+from cleveragents.config.settings import Settings
+from cleveragents.domain.models.acms.tiers import (
+    ActorContextView,
+    ActorRole,
+    ContextTier,
+    ScopedBackendView,
+    TierBudget,
+    TieredFragment,
+    TierMetrics,
+)
+
+__all__: list[str] = []
+
+
+# ---------------------------------------------------------------------------
+# ContextTier enum
+# ---------------------------------------------------------------------------
+
+
+@then('ContextTier should have values "hot", "warm", "cold"')
+def step_then_context_tier_values(context: Any) -> None:
+    assert ContextTier.HOT == "hot"
+    assert ContextTier.WARM == "warm"
+    assert ContextTier.COLD == "cold"
+    assert len(ContextTier) == 3
+
+
+# ---------------------------------------------------------------------------
+# ActorRole enum
+# ---------------------------------------------------------------------------
+
+
+@then('ActorRole should have values "strategist", "executor", "reviewer"')
+def step_then_actor_role_values(context: Any) -> None:
+    assert ActorRole.STRATEGIST == "strategist"
+    assert ActorRole.EXECUTOR == "executor"
+    assert ActorRole.REVIEWER == "reviewer"
+    assert len(ActorRole) == 3
+
+
+# ---------------------------------------------------------------------------
+# TieredFragment
+# ---------------------------------------------------------------------------
+
+
+@given('a TieredFragment with id "{fid}" and content "{content}"')
+def step_given_tiered_fragment(context: Any, fid: str, content: str) -> None:
+    context.fragment = TieredFragment(fragment_id=fid, content=content)
+
+
+@then('the fragment tier should be "{tier}"')
+def step_then_fragment_tier(context: Any, tier: str) -> None:
+    assert context.fragment.tier == tier
+
+
+@then("the fragment token_count should be {count:d}")
+def step_then_fragment_token_count(context: Any, count: int) -> None:
+    assert context.fragment.token_count == count
+
+
+@then("the fragment access_count should be {count:d}")
+def step_then_fragment_access_count(context: Any, count: int) -> None:
+    assert context.fragment.access_count == count
+
+
+@then("creating a TieredFragment with empty fragment_id should raise ValidationError")
+def step_then_empty_fragment_id(context: Any) -> None:
+    try:
+        TieredFragment(fragment_id="", content="x")
+        raise AssertionError("Expected ValidationError")
+    except ValidationError:
+        pass
+
+
+@then(
+    "creating a TieredFragment with negative token_count should raise ValidationError"
+)
+def step_then_negative_token_count(context: Any) -> None:
+    try:
+        TieredFragment(fragment_id="f", content="x", token_count=-1)
+        raise AssertionError("Expected ValidationError")
+    except ValidationError:
+        pass
+
+
+# ---------------------------------------------------------------------------
+# TierBudget
+# ---------------------------------------------------------------------------
+
+
+@given("a default TierBudget")
+def step_given_default_tier_budget(context: Any) -> None:
+    context.budget = TierBudget()
+
+
+@then("max_tokens_hot should be {val:d}")
+def step_then_max_tokens_hot(context: Any, val: int) -> None:
+    assert context.budget.max_tokens_hot == val
+
+
+@then("max_decisions_warm should be {val:d}")
+def step_then_max_decisions_warm(context: Any, val: int) -> None:
+    assert context.budget.max_decisions_warm == val
+
+
+@then("max_decisions_cold should be {val:d}")
+def step_then_max_decisions_cold(context: Any, val: int) -> None:
+    assert context.budget.max_decisions_cold == val
+
+
+# ---------------------------------------------------------------------------
+# ActorContextView
+# ---------------------------------------------------------------------------
+
+
+@given('an ActorContextView for "{role}"')
+def step_given_actor_context_view(context: Any, role: str) -> None:
+    context.actor_view = ActorContextView(actor_role=ActorRole(role))
+
+
+@then('effective tiers should be "hot", "warm", "cold"')
+def step_then_all_tiers(context: Any) -> None:
+    tiers = context.actor_view.get_effective_tiers()
+    assert tiers == [ContextTier.HOT, ContextTier.WARM, ContextTier.COLD]
+
+
+@then('effective tiers should be "hot", "warm"')
+def step_then_hot_warm(context: Any) -> None:
+    tiers = context.actor_view.get_effective_tiers()
+    assert tiers == [ContextTier.HOT, ContextTier.WARM]
+
+
+@then('effective tiers should be "hot"')
+def step_then_hot_only(context: Any) -> None:
+    tiers = context.actor_view.get_effective_tiers()
+    assert tiers == [ContextTier.HOT]
+
+
+# ---------------------------------------------------------------------------
+# TierMetrics
+# ---------------------------------------------------------------------------
+
+
+@given("a default TierMetrics")
+def step_given_default_metrics(context: Any) -> None:
+    context.metrics = TierMetrics()
+
+
+@then("total_fragments should be {val:d}")
+def step_then_total_fragments(context: Any, val: int) -> None:
+    assert context.metrics.total_fragments == val
+
+
+@then("hot_hit_rate should be {val:g}")
+def step_then_hot_hit_rate(context: Any, val: float) -> None:
+    assert context.metrics.hot_hit_rate == val
+
+
+# ---------------------------------------------------------------------------
+# ScopedBackendView
+# ---------------------------------------------------------------------------
+
+
+@given('a ScopedBackendView for project "{proj}"')
+def step_given_scoped_view(context: Any, proj: str) -> None:
+    context.scoped_view = ScopedBackendView(allowed_projects=[proj])
+    if not hasattr(context, "view_fragments"):
+        context.view_fragments = {}
+
+
+@given('a TieredFragment with id "{fid}" and no project')
+def step_given_fragment_no_project(context: Any, fid: str) -> None:
+    if not hasattr(context, "view_fragments"):
+        context.view_fragments = {}
+    context.view_fragments[fid] = TieredFragment(
+        fragment_id=fid, content="x", project_name=""
+    )
+
+
+@given('a TieredFragment with id "{fid}" in project "{proj}"')
+def step_given_fragment_in_project(context: Any, fid: str, proj: str) -> None:
+    if not hasattr(context, "view_fragments"):
+        context.view_fragments = {}
+    context.view_fragments[fid] = TieredFragment(
+        fragment_id=fid, content="x", project_name=proj
+    )
+
+
+@then('fragment "{fid}" should be visible')
+def step_then_fragment_visible(context: Any, fid: str) -> None:
+    frag = context.view_fragments[fid]
+    assert context.scoped_view.is_visible(frag)
+
+
+@then('fragment "{fid}" should not be visible')
+def step_then_fragment_not_visible(context: Any, fid: str) -> None:
+    frag = context.view_fragments[fid]
+    assert not context.scoped_view.is_visible(frag)
+
+
+# ---------------------------------------------------------------------------
+# ContextTierService: store/get
+# ---------------------------------------------------------------------------
+
+
+@given("a ContextTierService")
+def step_given_tier_service(context: Any) -> None:
+    context.tier_service = ContextTierService()
+
+
+@when('I store a fragment with id "{fid}" content "{content}" in tier "{tier}"')
+def step_when_store_fragment(context: Any, fid: str, content: str, tier: str) -> None:
+    frag = TieredFragment(
+        fragment_id=fid,
+        content=content,
+        tier=ContextTier(tier),
+    )
+    context.tier_service.store(frag)
+
+
+@when(
+    'I store fragment "{fid}" with content "{content}" tier "{tier}" project "{proj}"'
+)
+def step_when_store_fragment_project(
+    context: Any, fid: str, content: str, tier: str, proj: str
+) -> None:
+    frag = TieredFragment(
+        fragment_id=fid,
+        content=content,
+        tier=ContextTier(tier),
+        project_name=proj,
+    )
+    context.tier_service.store(frag)
+
+
+@when('I store a long fragment "{fid}" in tier "{tier}"')
+def step_when_store_long_fragment(context: Any, fid: str, tier: str) -> None:
+    long_content = "x" * 300  # Longer than _COLD_SUMMARY_MAX_CHARS (200)
+    frag = TieredFragment(
+        fragment_id=fid,
+        content=long_content,
+        tier=ContextTier(tier),
+    )
+    context.tier_service.store(frag)
+
+
+@then('tier getting "{fid}" should return content "{content}"')
+def step_then_get_content(context: Any, fid: str, content: str) -> None:
+    result = context.tier_service.get(fid)
+    assert result is not None
+    assert result.content == content
+
+
+@then('tier getting "{fid}" should return None')
+def step_then_get_none(context: Any, fid: str) -> None:
+    assert context.tier_service.get(fid) is None
+
+
+# ---------------------------------------------------------------------------
+# ContextTierService: promote/demote
+# ---------------------------------------------------------------------------
+
+
+@when('I promote "{fid}"')
+def step_when_promote(context: Any, fid: str) -> None:
+    context.last_promoted = context.tier_service.promote(fid)
+
+
+@when('I demote "{fid}"')
+def step_when_demote(context: Any, fid: str) -> None:
+    context.last_demoted = context.tier_service.demote(fid)
+
+
+@then('"{fid}" should be in tier "{tier}"')
+def step_then_in_tier(context: Any, fid: str, tier: str) -> None:
+    result = context.tier_service.get(fid)
+    assert result is not None
+    assert result.tier == tier
+
+
+@then('promoting "{fid}" should return None')
+def step_then_promote_none(context: Any, fid: str) -> None:
+    assert context.tier_service.promote(fid) is None
+
+
+@then('demoting "{fid}" should return None')
+def step_then_demote_none(context: Any, fid: str) -> None:
+    assert context.tier_service.demote(fid) is None
+
+
+@then('"{fid}" content should be summarized')
+def step_then_content_summarized(context: Any, fid: str) -> None:
+    result = context.tier_service.get(fid)
+    assert result is not None
+    # Summarised content ends with "..." if original was long
+    assert result.content.endswith("...")
+
+
+# ---------------------------------------------------------------------------
+# ContextTierService: LRU eviction
+# ---------------------------------------------------------------------------
+
+
+@when('I evict {count:d} LRU fragments from "{tier}"')
+def step_when_evict(context: Any, count: int, tier: str) -> None:
+    context.evicted = context.tier_service.evict_lru(ContextTier(tier), count)
+
+
+@then('"{tier}" tier should have {count:d} fragment')
+def step_then_tier_count_singular(context: Any, tier: str, count: int) -> None:
+    metrics = context.tier_service.get_metrics()
+    if tier == "hot":
+        assert metrics.hot_count == count
+    elif tier == "warm":
+        assert metrics.warm_count == count
+    else:
+        assert metrics.cold_count == count
+
+
+@then('evicting {count:d} from "{tier}" should raise ValueError')
+def step_then_evict_raises(context: Any, count: int, tier: str) -> None:
+    try:
+        context.tier_service.evict_lru(ContextTier(tier), count)
+        raise AssertionError("Expected ValueError")
+    except ValueError:
+        pass
+
+
+# ---------------------------------------------------------------------------
+# ContextTierService: actor views
+# ---------------------------------------------------------------------------
+
+
+@then('strategist for project "{proj}" should see {count:d} fragments')
+def step_then_strategist_sees(context: Any, proj: str, count: int) -> None:
+    frags = context.tier_service.get_for_actor(
+        ActorRole.STRATEGIST, project_names=[proj]
+    )
+    assert len(frags) == count
+
+
+@then('executor for project "{proj}" should see {count:d} fragments')
+def step_then_executor_sees(context: Any, proj: str, count: int) -> None:
+    frags = context.tier_service.get_for_actor(ActorRole.EXECUTOR, project_names=[proj])
+    assert len(frags) == count
+
+
+@then('reviewer for project "{proj}" should see {count:d} fragment')
+def step_then_reviewer_sees(context: Any, proj: str, count: int) -> None:
+    frags = context.tier_service.get_for_actor(ActorRole.REVIEWER, project_names=[proj])
+    assert len(frags) == count
+
+
+# ---------------------------------------------------------------------------
+# ContextTierService: scoped view
+# ---------------------------------------------------------------------------
+
+
+@then('scoped view for "{proj}" should have {count:d} fragment')
+def step_then_scoped_count(context: Any, proj: str, count: int) -> None:
+    frags = context.tier_service.get_scoped_view([proj])
+    assert len(frags) == count
+
+
+# ---------------------------------------------------------------------------
+# ContextTierService: metrics
+# ---------------------------------------------------------------------------
+
+
+@when('I get fragment "{fid}"')
+def step_when_get_fragment(context: Any, fid: str) -> None:
+    context.tier_service.get(fid)
+
+
+@then("hot_hit_count should be {val:d}")
+def step_then_hot_hit_count(context: Any, val: int) -> None:
+    metrics = context.tier_service.get_metrics()
+    assert metrics.hot_hit_count == val
+
+
+@then("hot_miss_count should be {val:d}")
+def step_then_hot_miss_count(context: Any, val: int) -> None:
+    metrics = context.tier_service.get_metrics()
+    assert metrics.hot_miss_count == val
+
+
+# ---------------------------------------------------------------------------
+# DI Container
+# ---------------------------------------------------------------------------
+
+
+@given("the DI container is reset")
+def step_given_di_reset(context: Any) -> None:
+    reset_container()
+    context.container = get_container()
+
+
+@then("the container should provide a context_tier_service")
+def step_then_container_has_tier_service(context: Any) -> None:
+    context.resolved_tier_svc = context.container.context_tier_service()
+    assert context.resolved_tier_svc is not None
+
+
+@then("the context_tier_service should be a ContextTierService")
+def step_then_tier_svc_type(context: Any) -> None:
+    assert isinstance(context.resolved_tier_svc, ContextTierService)
+
+
+@then("resolving context_tier_service twice should return the same instance")
+def step_then_tier_svc_singleton(context: Any) -> None:
+    a = context.container.context_tier_service()
+    b = context.container.context_tier_service()
+    assert a is b
+
+
+# ---------------------------------------------------------------------------
+# Settings
+# ---------------------------------------------------------------------------
+
+
+@then("settings should have context_max_tokens_hot")
+def step_then_settings_hot(context: Any) -> None:
+    s = Settings()
+    assert hasattr(s, "context_max_tokens_hot")
+    assert s.context_max_tokens_hot >= 0
+
+
+@then("settings should have context_max_decisions_warm")
+def step_then_settings_warm(context: Any) -> None:
+    s = Settings()
+    assert hasattr(s, "context_max_decisions_warm")
+    assert s.context_max_decisions_warm >= 0
+
+
+@then("settings should have context_max_decisions_cold")
+def step_then_settings_cold(context: Any) -> None:
+    s = Settings()
+    assert hasattr(s, "context_max_decisions_cold")
+    assert s.context_max_decisions_cold >= 0

robot/context_tiers.robot

@@ -0,0 +1,65 @@
+*** Settings ***
+Documentation    Smoke tests for ACMS Context Tiers (hot/warm/cold)
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_context_tiers.py
+
+*** Test Cases ***
+Tier Model Construction
+    [Documentation]    Verify ContextTier, ActorRole, TieredFragment creation
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    models    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    context-tiers-models-ok
+
+Store And Retrieve Fragment
+    [Documentation]    Verify storing and retrieving a fragment from the tier service
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    store-get    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    context-tiers-store-get-ok
+
+Promote And Demote
+    [Documentation]    Verify tier promotion and demotion
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    promote-demote    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    context-tiers-promote-demote-ok
+
+LRU Eviction
+    [Documentation]    Verify LRU eviction from hot tier
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    eviction    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    context-tiers-eviction-ok
+
+Actor Views
+    [Documentation]    Verify per-actor filtered views
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    actor-views    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    context-tiers-actor-views-ok
+
+Project Scoping
+    [Documentation]    Verify ScopedBackendView filters by project
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    scoping    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    context-tiers-scoping-ok
+
+DI Container Resolution
+    [Documentation]    Verify DI container provides context_tier_service
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    di-resolution    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    context-tiers-di-ok

robot/helper_context_tiers.py

@@ -0,0 +1,216 @@
+"""Robot Framework helper for ACMS context tier smoke tests.
+
+Provides a CLI-style interface for Robot to invoke tier model creation,
+service operations, and DI resolution.  Exit code 0 = success, 1 = failure.
+
+Usage:
+    python robot/helper_context_tiers.py <command>
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+# Ensure the src directory is on the import path.
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+from cleveragents.application.container import (  # noqa: E402
+    get_container,
+    reset_container,
+)
+from cleveragents.application.services.context_tiers import (  # noqa: E402
+    ContextTierService,
+)
+from cleveragents.domain.models.acms.tiers import (  # noqa: E402
+    ActorContextView,
+    ActorRole,
+    ContextTier,
+    ScopedBackendView,
+    TierBudget,
+    TieredFragment,
+    TierMetrics,
+)
+
+
+def main() -> int:
+    """Entry point called by Robot Framework ``Run Process``."""
+    if len(sys.argv) < 2:
+        print("Usage: helper_context_tiers.py <command>")
+        return 1
+
+    command: str = sys.argv[1]
+
+    if command == "models":
+        try:
+            assert ContextTier.HOT == "hot"
+            assert ActorRole.STRATEGIST == "strategist"
+            frag = TieredFragment(fragment_id="f1", content="hello")
+            assert frag.tier == ContextTier.HOT
+            budget = TierBudget()
+            assert budget.max_tokens_hot == 8000
+            view = ActorContextView(actor_role=ActorRole.EXECUTOR)
+            assert len(view.get_effective_tiers()) == 2
+            metrics = TierMetrics()
+            assert metrics.total_fragments == 0
+            print("context-tiers-models-ok")
+            return 0
+        except Exception as exc:
+            print(f"context-tiers-models-fail: {exc}")
+            return 1
+
+    if command == "store-get":
+        try:
+            svc = ContextTierService()
+            frag = TieredFragment(
+                fragment_id="f1", content="test", tier=ContextTier.HOT
+            )
+            svc.store(frag)
+            got = svc.get("f1")
+            assert got is not None
+            assert got.content == "test"
+            assert svc.get("missing") is None
+            print("context-tiers-store-get-ok")
+            return 0
+        except Exception as exc:
+            print(f"context-tiers-store-get-fail: {exc}")
+            return 1
+
+    if command == "promote-demote":
+        try:
+            svc = ContextTierService()
+            frag = TieredFragment(
+                fragment_id="f1", content="data", tier=ContextTier.COLD
+            )
+            svc.store(frag)
+            promoted = svc.promote("f1")
+            assert promoted is not None
+            assert promoted.tier == ContextTier.WARM
+            promoted2 = svc.promote("f1")
+            assert promoted2 is not None
+            assert promoted2.tier == ContextTier.HOT
+            demoted = svc.demote("f1")
+            assert demoted is not None
+            assert demoted.tier == ContextTier.WARM
+            print("context-tiers-promote-demote-ok")
+            return 0
+        except Exception as exc:
+            print(f"context-tiers-promote-demote-fail: {exc}")
+            return 1
+
+    if command == "eviction":
+        try:
+            svc = ContextTierService()
+            for i in range(5):
+                svc.store(
+                    TieredFragment(
+                        fragment_id=f"e{i}",
+                        content=f"data-{i}",
+                        tier=ContextTier.HOT,
+                    )
+                )
+            evicted = svc.evict_lru(ContextTier.HOT, 3)
+            assert len(evicted) == 3
+            assert svc.get_metrics().hot_count == 2
+            print("context-tiers-eviction-ok")
+            return 0
+        except Exception as exc:
+            print(f"context-tiers-eviction-fail: {exc}")
+            return 1
+
+    if command == "actor-views":
+        try:
+            svc = ContextTierService()
+            svc.store(
+                TieredFragment(
+                    fragment_id="h1",
+                    content="hot",
+                    tier=ContextTier.HOT,
+                    project_name="p1",
+                )
+            )
+            svc.store(
+                TieredFragment(
+                    fragment_id="w1",
+                    content="warm",
+                    tier=ContextTier.WARM,
+                    project_name="p1",
+                )
+            )
+            svc.store(
+                TieredFragment(
+                    fragment_id="c1",
+                    content="cold",
+                    tier=ContextTier.COLD,
+                    project_name="p1",
+                )
+            )
+            strat = svc.get_for_actor(ActorRole.STRATEGIST, ["p1"])
+            assert len(strat) == 3
+            exec_frags = svc.get_for_actor(ActorRole.EXECUTOR, ["p1"])
+            assert len(exec_frags) == 2
+            rev = svc.get_for_actor(ActorRole.REVIEWER, ["p1"])
+            assert len(rev) == 1
+            print("context-tiers-actor-views-ok")
+            return 0
+        except Exception as exc:
+            print(f"context-tiers-actor-views-fail: {exc}")
+            return 1
+
+    if command == "scoping":
+        try:
+            svc = ContextTierService()
+            svc.store(
+                TieredFragment(
+                    fragment_id="s1",
+                    content="yes",
+                    tier=ContextTier.HOT,
+                    project_name="alpha",
+                )
+            )
+            svc.store(
+                TieredFragment(
+                    fragment_id="s2",
+                    content="no",
+                    tier=ContextTier.HOT,
+                    project_name="beta",
+                )
+            )
+            scoped = svc.get_scoped_view(["alpha"])
+            assert len(scoped) == 1
+            assert scoped[0].fragment_id == "s1"
+
+            # ScopedBackendView direct test
+            view = ScopedBackendView(allowed_projects=["alpha"])
+            frag_a = TieredFragment(fragment_id="a", content="x", project_name="alpha")
+            frag_b = TieredFragment(fragment_id="b", content="x", project_name="beta")
+            assert view.is_visible(frag_a)
+            assert not view.is_visible(frag_b)
+            print("context-tiers-scoping-ok")
+            return 0
+        except Exception as exc:
+            print(f"context-tiers-scoping-fail: {exc}")
+            return 1
+
+    if command == "di-resolution":
+        try:
+            reset_container()
+            container = get_container()
+            svc = container.context_tier_service()
+            assert isinstance(svc, ContextTierService)
+            svc2 = container.context_tier_service()
+            assert svc is svc2  # singleton
+            print("context-tiers-di-ok")
+            return 0
+        except Exception as exc:
+            print(f"context-tiers-di-fail: {exc}")
+            return 1
+
+    print(f"Unknown command: {command}")
+    return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/cleveragents/application/container.py

@@ -16,6 +16,9 @@ from cleveragents.application.services.autonomy_guardrail_service import (
 )
 from cleveragents.application.services.checkpoint_service import CheckpointService
 from cleveragents.application.services.context_service import ContextService
+from cleveragents.application.services.context_tiers import (
+    ContextTierService,
+)
 from cleveragents.application.services.decision_service import DecisionService
 from cleveragents.application.services.plan_lifecycle_service import (
     PlanLifecycleService,
@@ -299,6 +302,12 @@ class Container(containers.DeclarativeContainer):
         database_url=database_url,
     )
 
+    # Context Tier Service - Singleton so all callers share tier state
+    context_tier_service = providers.Singleton(
+        ContextTierService,
+        settings=settings,
+    )
+
     # Skeleton Compressor Service - stateless, Singleton is sufficient
     skeleton_compressor_service = providers.Singleton(
         SkeletonCompressorService,

src/cleveragents/application/services/context_tiers.py

@@ -0,0 +1,338 @@
+"""Context Tier Service for hot/warm/cold fragment management.
+
+The ``ContextTierService`` manages the lifecycle of context fragments
+across three storage tiers:
+
+- **Hot** — in-memory ``dict``, fastest access, token-budget limited.
+- **Warm** — SQLite-backed (stub), moderate latency, decision-count limited.
+- **Cold** — file-backed (stub), highest latency, summarised on demotion.
+
+Supports promotion (cold→warm→hot), demotion (hot→warm→cold) with a
+summarisation hook, LRU eviction, per-actor filtered views, and
+project-scoped isolation via ``ScopedBackendView``.
+
+Based on ``docs/specification.md`` ACMS tier sections and issue #208.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from cleveragents.config.settings import Settings
+from cleveragents.domain.models.acms.tiers import (
+    ActorContextView,
+    ActorRole,
+    ContextTier,
+    ScopedBackendView,
+    TierBudget,
+    TieredFragment,
+    TierMetrics,
+)
+
+# ---------------------------------------------------------------------------
+# Default budget when settings are not provided
+# ---------------------------------------------------------------------------
+
+_DEFAULT_MAX_TOKENS_HOT = 8000
+_DEFAULT_MAX_DECISIONS_WARM = 500
+_DEFAULT_MAX_DECISIONS_COLD = 5000
+
+# Maximum content length kept after cold-tier summarisation
+_COLD_SUMMARY_MAX_CHARS = 200
+
+
+class ContextTierService:
+    """Manage context fragments across hot/warm/cold tiers.
+
+    The service is designed for dependency injection via the DI
+    container.  All mutable state lives in the three tier stores
+    and the metrics counters.
+    """
+
+    def __init__(self, settings: Settings | None = None) -> None:
+        budget = _budget_from_settings(settings)
+        self._budget = budget
+
+        # --- tier stores ---------------------------------------------------
+        self._hot: dict[str, TieredFragment] = {}
+        self._warm: dict[str, TieredFragment] = {}
+        self._cold: dict[str, TieredFragment] = {}
+
+        # --- metrics -------------------------------------------------------
+        self._hot_hit = 0
+        self._hot_miss = 0
+        self._warm_hit = 0
+        self._warm_miss = 0
+
+    # ------------------------------------------------------------------
+    # Store
+    # ------------------------------------------------------------------
+
+    def store(self, fragment: TieredFragment) -> None:
+        """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.
+
+        Args:
+            fragment: The fragment to store.
+
+        Raises:
+            ValueError: If ``fragment.fragment_id`` is empty.
+        """
+        if not fragment.fragment_id:
+            raise ValueError("fragment_id must be non-empty")
+
+        # Remove from other tiers first
+        self._remove_from_all(fragment.fragment_id)
+
+        if fragment.tier == ContextTier.HOT:
+            self._hot[fragment.fragment_id] = fragment
+        elif fragment.tier == ContextTier.WARM:
+            self._warm[fragment.fragment_id] = fragment
+        else:
+            self._cold[fragment.fragment_id] = fragment
+
+    # ------------------------------------------------------------------
+    # Retrieve
+    # ------------------------------------------------------------------
+
+    def get(self, fragment_id: str) -> TieredFragment | None:
+        """Retrieve a fragment from any tier, updating access metadata.
+
+        Returns ``None`` when the fragment is not found.
+        """
+        if fragment_id in self._hot:
+            self._hot_hit += 1
+            frag = self._hot[fragment_id]
+            return self._touch(frag)
+
+        self._hot_miss += 1
+
+        if fragment_id in self._warm:
+            self._warm_hit += 1
+            frag = self._warm[fragment_id]
+            return self._touch(frag)
+
+        self._warm_miss += 1
+
+        if fragment_id in self._cold:
+            return self._touch(self._cold[fragment_id])
+
+        return None
+
+    # ------------------------------------------------------------------
+    # Actor views
+    # ------------------------------------------------------------------
+
+    def get_for_actor(
+        self,
+        actor_role: ActorRole,
+        project_names: list[str] | None = None,
+    ) -> list[TieredFragment]:
+        """Return fragments visible to *actor_role*, scoped to projects.
+
+        Uses ``ActorContextView`` default tiers for the role and
+        ``ScopedBackendView`` for project isolation (when project_names
+        is provided).
+        """
+        view = ActorContextView(actor_role=actor_role)
+        effective_tiers = view.get_effective_tiers()
+
+        candidates: list[TieredFragment] = []
+        for tier in effective_tiers:
+            store = self._store_for_tier(tier)
+            candidates.extend(store.values())
+
+        # Apply project scoping
+        if project_names:
+            scoped = ScopedBackendView(allowed_projects=project_names)
+            candidates = [f for f in candidates if scoped.is_visible(f)]
+
+        # Sort by last_accessed descending, cap at max_fragments
+        candidates.sort(key=lambda f: f.last_accessed, reverse=True)
+        return candidates[: view.max_fragments]
+
+    # ------------------------------------------------------------------
+    # Scoped view
+    # ------------------------------------------------------------------
+
+    def get_scoped_view(
+        self,
+        project_names: list[str],
+    ) -> list[TieredFragment]:
+        """Return all fragments scoped to the given project names.
+
+        Enforces project isolation via ``ScopedBackendView`` so actors
+        never see resources outside the plan's project set.
+        """
+        if not project_names:
+            return []
+        scoped = ScopedBackendView(allowed_projects=project_names)
+        all_frags: list[TieredFragment] = []
+        for store in (self._hot, self._warm, self._cold):
+            all_frags.extend(f for f in store.values() if scoped.is_visible(f))
+        all_frags.sort(key=lambda f: f.last_accessed, reverse=True)
+        return all_frags
+
+    # ------------------------------------------------------------------
+    # Promote / Demote
+    # ------------------------------------------------------------------
+
+    def promote(self, fragment_id: str) -> TieredFragment | None:
+        """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.
+        """
+        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
+            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
+            return promoted
+
+        return None
+
+    def demote(self, fragment_id: str) -> TieredFragment | None:
+        """Demote a fragment one tier down (hot→warm or warm→cold).
+
+        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.
+        """
+        if fragment_id in self._hot:
+            frag = self._hot.pop(fragment_id)
+            demoted = frag.model_copy(update={"tier": ContextTier.WARM})
+            self._warm[fragment_id] = demoted
+            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})
+            self._cold[fragment_id] = demoted
+            return demoted
+
+        return None
+
+    # ------------------------------------------------------------------
+    # LRU eviction
+    # ------------------------------------------------------------------
+
+    def evict_lru(self, tier: ContextTier, count: int) -> list[str]:
+        """Evict the *count* least-recently-used fragments from *tier*.
+
+        Returns the list of evicted fragment IDs.
+
+        Raises:
+            ValueError: If *count* is not positive.
+        """
+        if count <= 0:
+            raise ValueError(f"count must be positive, got {count}")
+
+        store = self._store_for_tier(tier)
+        if not store:
+            return []
+
+        # Sort by last_accessed ascending (oldest first)
+        sorted_ids = sorted(
+            store.keys(),
+            key=lambda fid: store[fid].last_accessed,
+        )
+        to_evict = sorted_ids[:count]
+        for fid in to_evict:
+            del store[fid]
+        return to_evict
+
+    # ------------------------------------------------------------------
+    # Metrics
+    # ------------------------------------------------------------------
+
+    def get_metrics(self) -> TierMetrics:
+        """Return current hit/miss and population metrics."""
+        return TierMetrics(
+            hot_count=len(self._hot),
+            warm_count=len(self._warm),
+            cold_count=len(self._cold),
+            hot_hit_count=self._hot_hit,
+            hot_miss_count=self._hot_miss,
+            warm_hit_count=self._warm_hit,
+            warm_miss_count=self._warm_miss,
+        )
+
+    # ------------------------------------------------------------------
+    # Budget
+    # ------------------------------------------------------------------
+
+    @property
+    def budget(self) -> TierBudget:
+        """Return the active tier budget."""
+        return self._budget
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+
+    def _store_for_tier(
+        self,
+        tier: ContextTier,
+    ) -> dict[str, TieredFragment]:
+        """Return the backing store dict for *tier*."""
+        if tier == ContextTier.HOT:
+            return self._hot
+        if tier == ContextTier.WARM:
+            return self._warm
+        return self._cold
+
+    def _remove_from_all(self, fragment_id: str) -> None:
+        """Remove *fragment_id* from all tiers."""
+        self._hot.pop(fragment_id, None)
+        self._warm.pop(fragment_id, None)
+        self._cold.pop(fragment_id, None)
+
+    @staticmethod
+    def _touch(fragment: TieredFragment) -> TieredFragment:
+        """Update access metadata on *fragment* and return it."""
+        fragment.last_accessed = datetime.now(tz=UTC)
+        fragment.access_count += 1
+        return fragment
+
+    @staticmethod
+    def _summarize_for_cold(fragment: TieredFragment) -> TieredFragment:
+        """Summarisation hook: truncate content for cold storage.
+
+        This is a stub implementation that truncates the content.
+        A production implementation would call an LLM summariser.
+        """
+        content = fragment.content
+        if len(content) > _COLD_SUMMARY_MAX_CHARS:
+            content = content[:_COLD_SUMMARY_MAX_CHARS] + "..."
+        return fragment.model_copy(update={"content": content})
+
+
+# ---------------------------------------------------------------------------
+# Budget helper
+# ---------------------------------------------------------------------------
+
+
+def _budget_from_settings(settings: Settings | None) -> TierBudget:
+    """Build a ``TierBudget`` from application settings (or defaults)."""
+    if settings is None:
+        return TierBudget()
+    return TierBudget(
+        max_tokens_hot=getattr(
+            settings, "context_max_tokens_hot", _DEFAULT_MAX_TOKENS_HOT
+        ),
+        max_decisions_warm=getattr(
+            settings, "context_max_decisions_warm", _DEFAULT_MAX_DECISIONS_WARM
+        ),
+        max_decisions_cold=getattr(
+            settings, "context_max_decisions_cold", _DEFAULT_MAX_DECISIONS_COLD
+        ),
+    )

src/cleveragents/config/settings.py

@@ -225,6 +225,26 @@ class Settings(BaseSettings):
         validation_alias=AliasChoices("CLEVERAGENTS_TEST_DATABASE_URL"),
     )
 
+    # Context tier budgets (ACMS #208)
+    context_max_tokens_hot: int = Field(
+        default=8000,
+        ge=0,
+        validation_alias=AliasChoices("CLEVERAGENTS_CONTEXT_MAX_TOKENS_HOT"),
+        description="Maximum total tokens in the hot context tier.",
+    )
+    context_max_decisions_warm: int = Field(
+        default=500,
+        ge=0,
+        validation_alias=AliasChoices("CLEVERAGENTS_CONTEXT_MAX_DECISIONS_WARM"),
+        description="Maximum fragments in the warm context tier.",
+    )
+    context_max_decisions_cold: int = Field(
+        default=5000,
+        ge=0,
+        validation_alias=AliasChoices("CLEVERAGENTS_CONTEXT_MAX_DECISIONS_COLD"),
+        description="Maximum fragments in the cold context tier.",
+    )
+
     # Vector store configuration
     vector_store_enabled: bool = Field(
         default=False,

src/cleveragents/domain/models/acms/__init__.py

@@ -22,6 +22,15 @@ Stub backends (from :mod:`~cleveragents.domain.models.acms.stubs`):
 - ``InMemoryVectorBackend`` -- Zero-dependency vector search stub
 - ``InMemoryGraphBackend`` -- Zero-dependency graph query stub
 
+Tier types (from :mod:`~cleveragents.domain.models.acms.tiers`):
+- ``ContextTier`` -- Hot/warm/cold storage tier enumeration
+- ``ActorRole`` -- Strategist/executor/reviewer actor roles
+- ``TieredFragment`` -- Fragment with tier and access metadata
+- ``TierBudget`` -- Per-tier token/decision budget limits
+- ``ActorContextView`` -- Per-actor filtered view configuration
+- ``TierMetrics`` -- Hit/miss counters for cache monitoring
+- ``ScopedBackendView`` -- Project-scoped resource isolation
+
 Based on ``docs/specification.md`` ACMS / CRP sections and ADR-014.
 """
 
@@ -48,12 +57,24 @@ from cleveragents.domain.models.acms.stubs import (
     InMemoryTextBackend,
     InMemoryVectorBackend,
 )
+from cleveragents.domain.models.acms.tiers import (
+    ActorContextView,
+    ActorRole,
+    ContextTier,
+    ScopedBackendView,
+    TierBudget,
+    TieredFragment,
+    TierMetrics,
+)
 
 __all__: list[str] = [
+    "ActorContextView",
+    "ActorRole",
     "AssembledContext",
     "ContextBudget",
     "ContextFragment",
     "ContextRequest",
+    "ContextTier",
     "DetailLevelMap",
     "FragmentProvenance",
     "GraphBackend",
@@ -61,8 +82,12 @@ __all__: list[str] = [
     "InMemoryGraphBackend",
     "InMemoryTextBackend",
     "InMemoryVectorBackend",
+    "ScopedBackendView",
     "TextBackend",
     "TextResult",
+    "TierBudget",
+    "TierMetrics",
+    "TieredFragment",
     "VectorBackend",
     "VectorResult",
 ]

src/cleveragents/domain/models/acms/tiers.py

@@ -0,0 +1,289 @@
+"""Hot/warm/cold context tier models and actor view definitions.
+
+Provides the tiered context storage data types used by the ACMS
+Context Tier Service for managing context fragment lifecycle:
+
+| Type                | Role                                                   |
+|---------------------|--------------------------------------------------------|
+| ``ContextTier``     | Enumeration of hot/warm/cold storage tiers             |
+| ``ActorRole``       | Enumeration of actor roles (strategist/executor/reviewer)|
+| ``TieredFragment``  | A context fragment with tier and access metadata       |
+| ``TierBudget``      | Token/decision budget limits per tier                  |
+| ``ActorContextView``| Per-actor filtered view configuration                  |
+| ``TierMetrics``     | Hit/miss counters for cache performance monitoring     |
+| ``ScopedBackendView``| Project-scoped filter for resource isolation          |
+
+Based on ``docs/specification.md`` ACMS tier sections and issue #208.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from enum import StrEnum
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+# ---------------------------------------------------------------------------
+# Tier enumeration
+# ---------------------------------------------------------------------------
+
+
+class ContextTier(StrEnum):
+    """Storage tier for context fragments.
+
+    - ``HOT``  — in-memory, lowest latency, limited capacity
+    - ``WARM`` — SQLite-backed, moderate latency, larger capacity
+    - ``COLD`` — file-backed, highest latency, unlimited capacity
+    """
+
+    HOT = "hot"
+    WARM = "warm"
+    COLD = "cold"
+
+
+# ---------------------------------------------------------------------------
+# Actor role enumeration
+# ---------------------------------------------------------------------------
+
+
+class ActorRole(StrEnum):
+    """Actor roles that determine context visibility.
+
+    Each role has a default set of visible tiers and filtering rules.
+    """
+
+    STRATEGIST = "strategist"
+    EXECUTOR = "executor"
+    REVIEWER = "reviewer"
+
+
+# ---------------------------------------------------------------------------
+# TieredFragment
+# ---------------------------------------------------------------------------
+
+
+def _utc_now() -> datetime:
+    """Return the current UTC timestamp."""
+    return datetime.now(tz=UTC)
+
+
+class TieredFragment(BaseModel):
+    """A context fragment annotated with tier placement and access tracking.
+
+    Extends the conceptual ``ContextFragment`` from the CRP with
+    tier-specific metadata for LRU eviction and promotion/demotion
+    decisions.
+    """
+
+    fragment_id: str = Field(
+        ...,
+        min_length=1,
+        description="Stable unique identifier for this fragment",
+    )
+    content: str = Field(
+        ...,
+        description="Rendered text content of the fragment",
+    )
+    tier: ContextTier = Field(
+        default=ContextTier.HOT,
+        description="Current storage tier",
+    )
+    resource_id: str = Field(
+        default="",
+        description="Originating resource identifier",
+    )
+    project_name: str = Field(
+        default="",
+        description="Project scope for isolation filtering",
+    )
+    token_count: int = Field(
+        default=0,
+        ge=0,
+        description="Token count of content",
+    )
+    last_accessed: datetime = Field(
+        default_factory=_utc_now,
+        description="Timestamp of last access for LRU eviction",
+    )
+    access_count: int = Field(
+        default=0,
+        ge=0,
+        description="Number of times this fragment has been accessed",
+    )
+    created_at: datetime = Field(
+        default_factory=_utc_now,
+        description="Timestamp when the fragment was first stored",
+    )
+    metadata: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Arbitrary tier-specific metadata",
+    )
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+
+# ---------------------------------------------------------------------------
+# TierBudget
+# ---------------------------------------------------------------------------
+
+
+class TierBudget(BaseModel):
+    """Token and decision budget limits for each tier.
+
+    Controls how many tokens can reside in the hot tier, and how
+    many decisions (fragments) can be stored in warm and cold tiers.
+    """
+
+    max_tokens_hot: int = Field(
+        default=8000,
+        ge=0,
+        description="Maximum total tokens in the hot tier",
+    )
+    max_decisions_warm: int = Field(
+        default=500,
+        ge=0,
+        description="Maximum number of fragments in the warm tier",
+    )
+    max_decisions_cold: int = Field(
+        default=5000,
+        ge=0,
+        description="Maximum number of fragments in the cold tier",
+    )
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+
+# ---------------------------------------------------------------------------
+# ActorContextView
+# ---------------------------------------------------------------------------
+
+# Default visible tiers per actor role
+_DEFAULT_VISIBLE_TIERS: dict[ActorRole, list[ContextTier]] = {
+    ActorRole.STRATEGIST: [ContextTier.HOT, ContextTier.WARM, ContextTier.COLD],
+    ActorRole.EXECUTOR: [ContextTier.HOT, ContextTier.WARM],
+    ActorRole.REVIEWER: [ContextTier.HOT],
+}
+
+
+class ActorContextView(BaseModel):
+    """Configuration for a per-actor filtered context view.
+
+    Determines which tiers an actor can see and optionally restricts
+    results to specific projects.
+    """
+
+    actor_role: ActorRole = Field(
+        ...,
+        description="The role of the actor requesting context",
+    )
+    visible_tiers: list[ContextTier] = Field(
+        default_factory=list,
+        description="Tiers this actor can see (empty = use role defaults)",
+    )
+    project_filter: list[str] | None = Field(
+        default=None,
+        description="Restrict to these project names (None = no filter)",
+    )
+    max_fragments: int = Field(
+        default=100,
+        ge=1,
+        description="Maximum fragments to return in a single view",
+    )
+
+    @field_validator("visible_tiers", mode="before")
+    @classmethod
+    def _default_visible_tiers(
+        cls: type[ActorContextView],
+        v: list[ContextTier],
+    ) -> list[ContextTier]:
+        """Populate visible_tiers from role defaults when empty."""
+        # Actual default population happens in get_effective_tiers()
+        return v
+
+    def get_effective_tiers(self) -> list[ContextTier]:
+        """Return the effective visible tiers (role defaults if empty)."""
+        if self.visible_tiers:
+            return list(self.visible_tiers)
+        return list(_DEFAULT_VISIBLE_TIERS.get(self.actor_role, [ContextTier.HOT]))
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+
+# ---------------------------------------------------------------------------
+# TierMetrics
+# ---------------------------------------------------------------------------
+
+
+class TierMetrics(BaseModel):
+    """Hit/miss performance counters for the tier cache system."""
+
+    hot_count: int = Field(default=0, ge=0, description="Fragments in hot tier")
+    warm_count: int = Field(default=0, ge=0, description="Fragments in warm tier")
+    cold_count: int = Field(default=0, ge=0, description="Fragments in cold tier")
+    hot_hit_count: int = Field(default=0, ge=0, description="Hot tier cache hits")
+    hot_miss_count: int = Field(default=0, ge=0, description="Hot tier cache misses")
+    warm_hit_count: int = Field(default=0, ge=0, description="Warm tier cache hits")
+    warm_miss_count: int = Field(default=0, ge=0, description="Warm tier cache misses")
+
+    @property
+    def total_fragments(self) -> int:
+        """Total fragments across all tiers."""
+        return self.hot_count + self.warm_count + self.cold_count
+
+    @property
+    def hot_hit_rate(self) -> float:
+        """Hot tier hit rate (0.0 if no accesses)."""
+        total = self.hot_hit_count + self.hot_miss_count
+        if total == 0:
+            return 0.0
+        return self.hot_hit_count / total
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+
+# ---------------------------------------------------------------------------
+# ScopedBackendView
+# ---------------------------------------------------------------------------
+
+
+class ScopedBackendView(BaseModel):
+    """Project-scoped filter ensuring actors never see resources outside scope.
+
+    Enforces project isolation by restricting fragment visibility to
+    only those fragments whose ``project_name`` matches one of the
+    allowed project names.
+    """
+
+    allowed_projects: list[str] = Field(
+        ...,
+        min_length=1,
+        description="Project names the actor is permitted to access",
+    )
+
+    def is_visible(self, fragment: TieredFragment) -> bool:
+        """Return True if *fragment* belongs to an allowed project.
+
+        Fragments with an empty ``project_name`` are **excluded** by
+        default to prevent information leakage.
+        """
+        if not fragment.project_name:
+            return False
+        return fragment.project_name in self.allowed_projects
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )

vulture_whitelist.py

@@ -564,3 +564,33 @@ similarity_search  # noqa: B018, F821
 sparql_query  # noqa: B018, F821
 get_triples  # noqa: B018, F821
 traverse  # noqa: B018, F821
+
+# Context Tiers — public API (M6 ACMS, issue #208)
+ContextTier  # noqa: B018, F821
+ActorRole  # noqa: B018, F821
+TieredFragment  # noqa: B018, F821
+TierBudget  # noqa: B018, F821
+ActorContextView  # noqa: B018, F821
+TierMetrics  # noqa: B018, F821
+ScopedBackendView  # noqa: B018, F821
+ContextTierService  # noqa: B018, F821
+context_tier_service  # noqa: B018, F821
+context_max_tokens_hot  # noqa: B018, F821
+context_max_decisions_warm  # noqa: B018, F821
+context_max_decisions_cold  # noqa: B018, F821
+get_effective_tiers  # noqa: B018, F821
+total_fragments  # noqa: B018, F821
+hot_hit_rate  # noqa: B018, F821
+is_visible  # noqa: B018, F821
+get_for_actor  # noqa: B018, F821
+get_scoped_view  # noqa: B018, F821
+get_metrics  # noqa: B018, F821
+evict_lru  # noqa: B018, F821
+_summarize_for_cold  # noqa: B018, F821
+_budget_from_settings  # noqa: B018, F821
+_DEFAULT_VISIBLE_TIERS  # noqa: B018, F821
+_COLD_SUMMARY_MAX_CHARS  # noqa: B018, F821
+_DEFAULT_MAX_TOKENS_HOT  # noqa: B018, F821
+_DEFAULT_MAX_DECISIONS_WARM  # noqa: B018, F821
+_DEFAULT_MAX_DECISIONS_COLD  # noqa: B018, F821
+_utc_now  # noqa: B018, F821