features/acms_context_analysis_engine.feature

@@ -0,0 +1,195 @@
+Feature: ACMS Context Analysis Engine
+  As a CleverAgents user
+  I want to analyze the ACMS context index
+  So that I can understand how my context budget is being used
+
+  # ── entry_count ────────────────────────────────────────────
+
+  Scenario: entry_count returns zero for empty index
+    Given an empty ContextAnalysisEngine
+    When I call entry_count
+    Then the entry count should be 0
+
+  Scenario: entry_count returns total across all tiers
+    Given a ContextAnalysisEngine with fragments in all tiers
+    When I call entry_count
+    Then the entry count should be 3
+
+  # ── tier_distribution ──────────────────────────────────────
+
+  Scenario: tier_distribution returns zero counts for empty index
+    Given an empty ContextAnalysisEngine
+    When I call tier_distribution
+    Then the hot tier count should be 0
+    And the warm tier count should be 0
+    And the cold tier count should be 0
+
+  Scenario: tier_distribution counts fragments per tier
+    Given a ContextAnalysisEngine with one hot fragment of content "hello"
+    When I call tier_distribution
+    Then the hot tier count should be 1
+    And the hot tier size_bytes should be 5
+
+  Scenario: tier_distribution counts warm fragments
+    Given a ContextAnalysisEngine with one warm fragment of content "world"
+    When I call tier_distribution
+    Then the warm tier count should be 1
+    And the warm tier size_bytes should be 5
+
+  Scenario: tier_distribution counts cold fragments
+    Given a ContextAnalysisEngine with one cold fragment of content "cold"
+    When I call tier_distribution
+    Then the cold tier count should be 1
+    And the cold tier size_bytes should be 4
+
+  Scenario: tier_distribution aggregates sizes across multiple fragments
+    Given a ContextAnalysisEngine with two hot fragments of content "ab" and "cde"
+    When I call tier_distribution
+    Then the hot tier count should be 2
+    And the hot tier size_bytes should be 5
+
+  # ── budget_utilization ─────────────────────────────────────
+
+  Scenario: budget_utilization returns zero for empty index
+    Given an empty ContextAnalysisEngine with max_total_size 1000
+    When I call budget_utilization
+    Then the current_bytes should be 0
+    And the max_bytes should be 1000
+    And the utilization_pct should be 0.0
+
+  Scenario: budget_utilization computes percentage correctly
+    Given a ContextAnalysisEngine with a hot fragment of content "hello" and max_total_size 10
+    When I call budget_utilization
+    Then the current_bytes should be 5
+    And the max_bytes should be 10
+    And the utilization_pct should be 50.0
+
+  Scenario: budget_utilization caps at 100 percent when over budget
+    Given a ContextAnalysisEngine with a hot fragment of content "hello world" and max_total_size 5
+    When I call budget_utilization
+    Then the utilization_pct should be 100.0
+
+  Scenario: budget_utilization returns zero when max_bytes is zero
+    Given an empty ContextAnalysisEngine with max_total_size 0
+    When I call budget_utilization
+    Then the utilization_pct should be 0.0
+
+  # ── top_files ──────────────────────────────────────────────
+
+  Scenario: top_files returns empty list for empty index
+    Given an empty ContextAnalysisEngine
+    When I call top_files with n 10
+    Then the top files list should be empty
+
+  Scenario: top_files returns entries sorted by access_count descending
+    Given a ContextAnalysisEngine with fragments having access counts 5 and 2 and 8
+    When I call top_files with n 10
+    Then the top files should be ordered by access_count descending
+
+  Scenario: top_files respects the n limit
+    Given a ContextAnalysisEngine with fragments having access counts 5 and 2 and 8
+    When I call top_files with n 2
+    Then the top files list should have 2 entries
+
+  Scenario: top_files raises ValueError for non-positive n
+    Given an empty ContextAnalysisEngine
+    When I call top_files with n 0
+    Then a ValueError should be raised for top_files n
+
+  Scenario: top_files includes fragment_id resource_id access_count and tier
+    Given a ContextAnalysisEngine with one hot fragment with resource_id "uko:file/main.py" and access_count 3
+    When I call top_files with n 10
+    Then the first top file should have resource_id "uko:file/main.py"
+    And the first top file should have access_count 3
+    And the first top file should have tier "hot"
+
+  # ── analyze ────────────────────────────────────────────────
+
+  Scenario: analyze returns combined AnalysisResult
+    Given a ContextAnalysisEngine with fragments in all tiers
+    When I call analyze with top_n 10
+    Then the analysis entry_count should be 3
+    And the analysis tier_distribution should have hot count 1
+    And the analysis top_files should not be empty
+
+  # ── format_json ────────────────────────────────────────────
+
+  Scenario: format_json returns valid JSON with all keys
+    Given a ContextAnalysisEngine with fragments in all tiers
+    When I call analyze with top_n 10
+    And I format the result as JSON
+    Then the acms JSON output should contain key "entry_count"
+    And the acms JSON output should contain key "tier_distribution"
+    And the acms JSON output should contain key "budget_utilization"
+    And the acms JSON output should contain key "top_files"
+
+  Scenario: format_json tier_distribution has hot warm cold keys
+    Given a ContextAnalysisEngine with fragments in all tiers
+    When I call analyze with top_n 10
+    And I format the result as JSON
+    Then the acms JSON tier_distribution should have keys "hot" "warm" "cold"
+
+  Scenario: format_json budget_utilization has required keys
+    Given a ContextAnalysisEngine with fragments in all tiers
+    When I call analyze with top_n 10
+    And I format the result as JSON
+    Then the acms JSON budget_utilization should have keys "current_bytes" "max_bytes" "utilization_pct"
+
+  # ── format_text ────────────────────────────────────────────
+
+  Scenario: format_text returns human-readable output
+    Given a ContextAnalysisEngine with fragments in all tiers
+    When I call analyze with top_n 10
+    And I format the result as text
+    Then the text output should contain "ACMS Context Analysis"
+    And the text output should contain "Tier Distribution"
+    And the text output should contain "Budget Utilization"
+    And the text output should contain "Top"
+
+  Scenario: format_text shows no entries when index is empty
+    Given an empty ContextAnalysisEngine
+    When I call analyze with top_n 10
+    And I format the result as text
+    Then the text output should contain "(no entries)"
+
+  # ── to_dict ────────────────────────────────────────────────
+
+  Scenario: AnalysisResult to_dict contains all keys
+    Given a ContextAnalysisEngine with fragments in all tiers
+    When I call analyze with top_n 10
+    Then the result to_dict should contain key "entry_count"
+    And the result to_dict should contain key "tier_distribution"
+    And the result to_dict should contain key "budget_utilization"
+    And the result to_dict should contain key "top_files"
+
+  Scenario: TierStats to_dict returns count and size_bytes
+    Given a TierStats with count 3 and size_bytes 100
+    When I call to_dict on TierStats
+    Then the TierStats dict should have count 3 and size_bytes 100
+
+  Scenario: BudgetUtilization to_dict rounds utilization_pct
+    Given a BudgetUtilization with current 50 max 100 pct 50.123456
+    When I call to_dict on BudgetUtilization
+    Then the BudgetUtilization dict utilization_pct should be 50.12
+
+  Scenario: TopFileEntry to_dict returns all fields
+    Given a TopFileEntry with fragment_id "f1" resource_id "r1" access_count 5 tier "hot"
+    When I call to_dict on TopFileEntry
+    Then the TopFileEntry dict should have all fields
+
+  # ── default max_total_size ─────────────────────────────────
+
+  Scenario: engine uses hot-tier budget as default max_total_size
+    Given an empty ContextAnalysisEngine without explicit max_total_size
+    When I call budget_utilization
+    Then the max_bytes should be the hot-tier budget
+
+  # ── CLI analyze command ────────────────────────────────────
+
+  Scenario: context analyze CLI command produces text output
+    When I invoke the context analyze CLI command with empty tier service
+    Then the CLI output should contain "ACMS Context Analysis"
+
+  Scenario: context analyze CLI command produces JSON output with --format json
+    When I invoke the context analyze CLI command with empty tier service and format json
+    Then the acms CLI JSON output should contain key "entry_count"

features/steps/acms_context_analysis_engine_steps.py

@@ -0,0 +1,592 @@
+"""Step definitions for acms_context_analysis_engine.feature."""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from behave import given, then, when  # type: ignore[import-untyped]
+from behave.runner import Context  # type: ignore[import-untyped]
+
+from cleveragents.application.services.context_analysis_engine import (
+    AnalysisResult,
+    BudgetUtilization,
+    ContextAnalysisEngine,
+    TierDistribution,
+    TierStats,
+    TopFileEntry,
+)
+from cleveragents.application.services.context_tiers import ContextTierService
+from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_fragment(
+    fragment_id: str,
+    content: str,
+    tier: ContextTier,
+    resource_id: str = "",
+    access_count: int = 0,
+) -> TieredFragment:
+    return TieredFragment(
+        fragment_id=fragment_id,
+        content=content,
+        tier=tier,
+        resource_id=resource_id,
+        access_count=access_count,
+    )
+
+
+def _make_engine(
+    tier_service: ContextTierService,
+    max_total_size: int | None = None,
+) -> ContextAnalysisEngine:
+    return ContextAnalysisEngine(
+        tier_service=tier_service,
+        max_total_size=max_total_size,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Given steps
+# ---------------------------------------------------------------------------
+
+
+@given("an empty ContextAnalysisEngine")
+def step_empty_engine(context: Context) -> None:
+    context.tier_service = ContextTierService()
+    context.engine = _make_engine(context.tier_service)
+
+
+@given("an empty ContextAnalysisEngine with max_total_size {size:d}")
+def step_empty_engine_with_max(context: Context, size: int) -> None:
+    context.tier_service = ContextTierService()
+    context.engine = _make_engine(context.tier_service, max_total_size=size)
+
+
+@given("an empty ContextAnalysisEngine without explicit max_total_size")
+def step_empty_engine_no_max(context: Context) -> None:
+    context.tier_service = ContextTierService()
+    context.engine = ContextAnalysisEngine(tier_service=context.tier_service)
+
+
+@given("a ContextAnalysisEngine with fragments in all tiers")
+def step_engine_all_tiers(context: Context) -> None:
+    context.tier_service = ContextTierService()
+    context.tier_service.store(
+        _make_fragment("hot-1", "hot content", ContextTier.HOT, access_count=5)
+    )
+    context.tier_service.store(
+        _make_fragment("warm-1", "warm content", ContextTier.WARM, access_count=2)
+    )
+    context.tier_service.store(
+        _make_fragment("cold-1", "cold content", ContextTier.COLD, access_count=1)
+    )
+    context.engine = _make_engine(context.tier_service, max_total_size=10000)
+
+
+@given('a ContextAnalysisEngine with one hot fragment of content "hello"')
+def step_engine_one_hot_hello(context: Context) -> None:
+    context.tier_service = ContextTierService()
+    context.tier_service.store(_make_fragment("hot-1", "hello", ContextTier.HOT))
+    context.engine = _make_engine(context.tier_service)
+
+
+@given('a ContextAnalysisEngine with one warm fragment of content "world"')
+def step_engine_one_warm_world(context: Context) -> None:
+    context.tier_service = ContextTierService()
+    context.tier_service.store(_make_fragment("warm-1", "world", ContextTier.WARM))
+    context.engine = _make_engine(context.tier_service)
+
+
+@given('a ContextAnalysisEngine with one cold fragment of content "cold"')
+def step_engine_one_cold_cold(context: Context) -> None:
+    context.tier_service = ContextTierService()
+    context.tier_service.store(_make_fragment("cold-1", "cold", ContextTier.COLD))
+    context.engine = _make_engine(context.tier_service)
+
+
+@given('a ContextAnalysisEngine with two hot fragments of content "ab" and "cde"')
+def step_engine_two_hot_ab_cde(context: Context) -> None:
+    context.tier_service = ContextTierService()
+    context.tier_service.store(_make_fragment("hot-1", "ab", ContextTier.HOT))
+    context.tier_service.store(_make_fragment("hot-2", "cde", ContextTier.HOT))
+    context.engine = _make_engine(context.tier_service)
+
+
+@given(
+    'a ContextAnalysisEngine with a hot fragment of content "hello" and max_total_size 10'
+)
+def step_engine_hot_hello_max10(context: Context) -> None:
+    context.tier_service = ContextTierService()
+    context.tier_service.store(_make_fragment("hot-1", "hello", ContextTier.HOT))
+    context.engine = _make_engine(context.tier_service, max_total_size=10)
+
+
+@given(
+    'a ContextAnalysisEngine with a hot fragment of content "hello world" and max_total_size 5'
+)
+def step_engine_hot_hello_world_max5(context: Context) -> None:
+    context.tier_service = ContextTierService()
+    context.tier_service.store(_make_fragment("hot-1", "hello world", ContextTier.HOT))
+    context.engine = _make_engine(context.tier_service, max_total_size=5)
+
+
+@given(
+    "a ContextAnalysisEngine with fragments having access counts {a:d} and {b:d} and {c:d}"
+)
+def step_engine_access_counts(context: Context, a: int, b: int, c: int) -> None:
+    context.tier_service = ContextTierService()
+    context.tier_service.store(
+        _make_fragment("frag-a", "content a", ContextTier.HOT, access_count=a)
+    )
+    context.tier_service.store(
+        _make_fragment("frag-b", "content b", ContextTier.WARM, access_count=b)
+    )
+    context.tier_service.store(
+        _make_fragment("frag-c", "content c", ContextTier.COLD, access_count=c)
+    )
+    context.engine = _make_engine(context.tier_service)
+
+
+@given(
+    'a ContextAnalysisEngine with one hot fragment with resource_id "uko:file/main.py" and access_count 3'
+)
+def step_engine_hot_resource_main(context: Context) -> None:
+    context.tier_service = ContextTierService()
+    context.tier_service.store(
+        _make_fragment(
+            "hot-1",
+            "content",
+            ContextTier.HOT,
+            resource_id="uko:file/main.py",
+            access_count=3,
+        )
+    )
+    context.engine = _make_engine(context.tier_service)
+
+
+@given("a TierStats with count {count:d} and size_bytes {size:d}")
+def step_tier_stats(context: Context, count: int, size: int) -> None:
+    context.tier_stats = TierStats(count=count, size_bytes=size)
+
+
+@given("a BudgetUtilization with current {current:d} max {max_b:d} pct {pct:f}")
+def step_budget_util(context: Context, current: int, max_b: int, pct: float) -> None:
+    context.budget_util = BudgetUtilization(
+        current_bytes=current,
+        max_bytes=max_b,
+        utilization_pct=pct,
+    )
+
+
+@given(
+    'a TopFileEntry with fragment_id "f1" resource_id "r1" access_count 5 tier "hot"'
+)
+def step_top_file_entry(context: Context) -> None:
+    context.top_file_entry = TopFileEntry(
+        fragment_id="f1",
+        resource_id="r1",
+        access_count=5,
+        tier="hot",
+    )
+
+
+# ---------------------------------------------------------------------------
+# When steps
+# ---------------------------------------------------------------------------
+
+
+@when("I call entry_count")
+def step_call_entry_count(context: Context) -> None:
+    context.result = context.engine.entry_count()
+
+
+@when("I call tier_distribution")
+def step_call_tier_distribution(context: Context) -> None:
+    context.result = context.engine.tier_distribution()
+
+
+@when("I call budget_utilization")
+def step_call_budget_utilization(context: Context) -> None:
+    context.result = context.engine.budget_utilization()
+
+
+@when("I call top_files with n {n:d}")
+def step_call_top_files(context: Context, n: int) -> None:
+    context.raised_error = None
+    try:
+        context.result = context.engine.top_files(n=n)
+    except ValueError as exc:
+        context.raised_error = exc
+
+
+@when("I call analyze with top_n {top_n:d}")
+def step_call_analyze(context: Context, top_n: int) -> None:
+    context.analysis_result = context.engine.analyze(top_n=top_n)
+
+
+@when("I format the result as JSON")
+def step_format_json(context: Context) -> None:
+    context.acms_json_output = ContextAnalysisEngine.format_json(
+        context.analysis_result
+    )
+    context.acms_json_data = json.loads(context.acms_json_output)
+
+
+@when("I format the result as text")
+def step_format_text(context: Context) -> None:
+    context.text_output = ContextAnalysisEngine.format_text(context.analysis_result)
+
+
+@when("I call to_dict on TierStats")
+def step_tier_stats_to_dict(context: Context) -> None:
+    context.result = context.tier_stats.to_dict()
+
+
+@when("I call to_dict on BudgetUtilization")
+def step_budget_util_to_dict(context: Context) -> None:
+    context.result = context.budget_util.to_dict()
+
+
+@when("I call to_dict on TopFileEntry")
+def step_top_file_to_dict(context: Context) -> None:
+    context.result = context.top_file_entry.to_dict()
+
+
+@when("I invoke the context analyze CLI command with empty tier service")
+def step_invoke_cli_analyze_empty(context: Context) -> None:
+    from typer.testing import CliRunner
+
+    from cleveragents.cli.commands.context import app
+
+    runner = CliRunner()
+    result = runner.invoke(app, ["analyze"])
+    context.cli_result = result
+    context.cli_output = result.output
+
+
+@when(
+    "I invoke the context analyze CLI command with empty tier service and format json"
+)
+def step_invoke_cli_analyze_empty_json(context: Context) -> None:
+    from typer.testing import CliRunner
+
+    from cleveragents.cli.commands.context import app
+
+    runner = CliRunner()
+    result = runner.invoke(app, ["analyze", "--format", "json"])
+    context.cli_result = result
+    context.cli_output = result.output
+    context.acms_cli_json_data = json.loads(result.output)
+
+
+# ---------------------------------------------------------------------------
+# Then steps
+# ---------------------------------------------------------------------------
+
+
+@then("the entry count should be {expected:d}")
+def step_assert_entry_count(context: Context, expected: int) -> None:
+    assert context.result == expected, (
+        f"Expected entry count {expected}, got {context.result}"
+    )
+
+
+@then("the hot tier count should be {expected:d}")
+def step_assert_hot_count(context: Context, expected: int) -> None:
+    dist: TierDistribution = context.result
+    assert dist.hot.count == expected, (
+        f"Expected hot count {expected}, got {dist.hot.count}"
+    )
+
+
+@then("the warm tier count should be {expected:d}")
+def step_assert_warm_count(context: Context, expected: int) -> None:
+    dist: TierDistribution = context.result
+    assert dist.warm.count == expected, (
+        f"Expected warm count {expected}, got {dist.warm.count}"
+    )
+
+
+@then("the cold tier count should be {expected:d}")
+def step_assert_cold_count(context: Context, expected: int) -> None:
+    dist: TierDistribution = context.result
+    assert dist.cold.count == expected, (
+        f"Expected cold count {expected}, got {dist.cold.count}"
+    )
+
+
+@then("the hot tier size_bytes should be {expected:d}")
+def step_assert_hot_size(context: Context, expected: int) -> None:
+    dist: TierDistribution = context.result
+    assert dist.hot.size_bytes == expected, (
+        f"Expected hot size_bytes {expected}, got {dist.hot.size_bytes}"
+    )
+
+
+@then("the warm tier size_bytes should be {expected:d}")
+def step_assert_warm_size(context: Context, expected: int) -> None:
+    dist: TierDistribution = context.result
+    assert dist.warm.size_bytes == expected, (
+        f"Expected warm size_bytes {expected}, got {dist.warm.size_bytes}"
+    )
+
+
+@then("the cold tier size_bytes should be {expected:d}")
+def step_assert_cold_size(context: Context, expected: int) -> None:
+    dist: TierDistribution = context.result
+    assert dist.cold.size_bytes == expected, (
+        f"Expected cold size_bytes {expected}, got {dist.cold.size_bytes}"
+    )
+
+
+@then("the current_bytes should be {expected:d}")
+def step_assert_current_bytes(context: Context, expected: int) -> None:
+    util: BudgetUtilization = context.result
+    assert util.current_bytes == expected, (
+        f"Expected current_bytes {expected}, got {util.current_bytes}"
+    )
+
+
+@then("the max_bytes should be {expected:d}")
+def step_assert_max_bytes(context: Context, expected: int) -> None:
+    util: BudgetUtilization = context.result
+    assert util.max_bytes == expected, (
+        f"Expected max_bytes {expected}, got {util.max_bytes}"
+    )
+
+
+@then("the utilization_pct should be {expected:f}")
+def step_assert_utilization_pct(context: Context, expected: float) -> None:
+    util: BudgetUtilization = context.result
+    assert abs(util.utilization_pct - expected) < 0.01, (
+        f"Expected utilization_pct {expected}, got {util.utilization_pct}"
+    )
+
+
+@then("the top files list should be empty")
+def step_assert_top_files_empty(context: Context) -> None:
+    assert context.result == [], f"Expected empty list, got {context.result}"
+
+
+@then("the top files should be ordered by access_count descending")
+def step_assert_top_files_ordered(context: Context) -> None:
+    files: list[TopFileEntry] = context.result
+    counts = [f.access_count for f in files]
+    assert counts == sorted(counts, reverse=True), (
+        f"Expected descending order, got {counts}"
+    )
+
+
+@then("the top files list should have {expected:d} entries")
+def step_assert_top_files_count(context: Context, expected: int) -> None:
+    assert len(context.result) == expected, (
+        f"Expected {expected} entries, got {len(context.result)}"
+    )
+
+
+@then("a ValueError should be raised for top_files n")
+def step_assert_value_error_top_files(context: Context) -> None:
+    assert context.raised_error is not None, "Expected ValueError but none was raised"
+    assert isinstance(context.raised_error, ValueError)
+
+
+@then('the first top file should have resource_id "uko:file/main.py"')
+def step_assert_first_resource_id(context: Context) -> None:
+    files: list[TopFileEntry] = context.result
+    assert len(files) > 0, "Expected at least one top file"
+    assert files[0].resource_id == "uko:file/main.py", (
+        f"Expected resource_id 'uko:file/main.py', got {files[0].resource_id!r}"
+    )
+
+
+@then("the first top file should have access_count {expected:d}")
+def step_assert_first_access_count(context: Context, expected: int) -> None:
+    files: list[TopFileEntry] = context.result
+    assert files[0].access_count == expected, (
+        f"Expected access_count {expected}, got {files[0].access_count}"
+    )
+
+
+@then('the first top file should have tier "hot"')
+def step_assert_first_tier_hot(context: Context) -> None:
+    files: list[TopFileEntry] = context.result
+    assert files[0].tier == "hot", f"Expected tier 'hot', got {files[0].tier!r}"
+
+
+@then("the analysis entry_count should be {expected:d}")
+def step_assert_analysis_entry_count(context: Context, expected: int) -> None:
+    result: AnalysisResult = context.analysis_result
+    assert result.entry_count == expected, (
+        f"Expected entry_count {expected}, got {result.entry_count}"
+    )
+
+
+@then("the analysis tier_distribution should have hot count {expected:d}")
+def step_assert_analysis_hot_count(context: Context, expected: int) -> None:
+    result: AnalysisResult = context.analysis_result
+    assert result.tier_distribution.hot.count == expected, (
+        f"Expected hot count {expected}, got {result.tier_distribution.hot.count}"
+    )
+
+
+@then("the analysis top_files should not be empty")
+def step_assert_analysis_top_files_not_empty(context: Context) -> None:
+    result: AnalysisResult = context.analysis_result
+    assert len(result.top_files) > 0, "Expected non-empty top_files"
+
+
+@then('the acms JSON output should contain key "entry_count"')
+def step_assert_acms_json_entry_count(context: Context) -> None:
+    assert "entry_count" in context.acms_json_data, (
+        "Expected key 'entry_count' in JSON output"
+    )
+
+
+@then('the acms JSON output should contain key "tier_distribution"')
+def step_assert_acms_json_tier_dist(context: Context) -> None:
+    assert "tier_distribution" in context.acms_json_data, (
+        "Expected key 'tier_distribution' in JSON output"
+    )
+
+
+@then('the acms JSON output should contain key "budget_utilization"')
+def step_assert_acms_json_budget(context: Context) -> None:
+    assert "budget_utilization" in context.acms_json_data, (
+        "Expected key 'budget_utilization' in JSON output"
+    )
+
+
+@then('the acms JSON output should contain key "top_files"')
+def step_assert_acms_json_top_files(context: Context) -> None:
+    assert "top_files" in context.acms_json_data, (
+        "Expected key 'top_files' in JSON output"
+    )
+
+
+@then('the acms JSON tier_distribution should have keys "hot" "warm" "cold"')
+def step_assert_acms_json_tier_keys(context: Context) -> None:
+    tier_dist = context.acms_json_data.get("tier_distribution", {})
+    for key in ("hot", "warm", "cold"):
+        assert key in tier_dist, (
+            f"Expected key {key!r} in tier_distribution, got: {list(tier_dist.keys())}"
+        )
+
+
+@then(
+    'the acms JSON budget_utilization should have keys "current_bytes" "max_bytes" "utilization_pct"'
+)
+def step_assert_acms_json_budget_keys(context: Context) -> None:
+    budget = context.acms_json_data.get("budget_utilization", {})
+    for key in ("current_bytes", "max_bytes", "utilization_pct"):
+        assert key in budget, (
+            f"Expected key {key!r} in budget_utilization, got: {list(budget.keys())}"
+        )
+
+
+@then('the text output should contain "ACMS Context Analysis"')
+def step_assert_text_acms(context: Context) -> None:
+    assert "ACMS Context Analysis" in context.text_output, (
+        f"Expected 'ACMS Context Analysis' in text output:\n{context.text_output}"
+    )
+
+
+@then('the text output should contain "Tier Distribution"')
+def step_assert_text_tier_dist(context: Context) -> None:
+    assert "Tier Distribution" in context.text_output, (
+        f"Expected 'Tier Distribution' in text output:\n{context.text_output}"
+    )
+
+
+@then('the text output should contain "Budget Utilization"')
+def step_assert_text_budget(context: Context) -> None:
+    assert "Budget Utilization" in context.text_output, (
+        f"Expected 'Budget Utilization' in text output:\n{context.text_output}"
+    )
+
+
+@then('the text output should contain "Top"')
+def step_assert_text_top(context: Context) -> None:
+    assert "Top" in context.text_output, (
+        f"Expected 'Top' in text output:\n{context.text_output}"
+    )
+
+
+@then('the text output should contain "(no entries)"')
+def step_assert_text_no_entries(context: Context) -> None:
+    assert "(no entries)" in context.text_output, (
+        f"Expected '(no entries)' in text output:\n{context.text_output}"
+    )
+
+
+@then('the result to_dict should contain key "entry_count"')
+def step_assert_result_dict_entry_count(context: Context) -> None:
+    d = context.analysis_result.to_dict()
+    assert "entry_count" in d, "Expected key 'entry_count' in result dict"
+
+
+@then('the result to_dict should contain key "tier_distribution"')
+def step_assert_result_dict_tier_dist(context: Context) -> None:
+    d = context.analysis_result.to_dict()
+    assert "tier_distribution" in d, "Expected key 'tier_distribution' in result dict"
+
+
+@then('the result to_dict should contain key "budget_utilization"')
+def step_assert_result_dict_budget(context: Context) -> None:
+    d = context.analysis_result.to_dict()
+    assert "budget_utilization" in d, "Expected key 'budget_utilization' in result dict"
+
+
+@then('the result to_dict should contain key "top_files"')
+def step_assert_result_dict_top_files(context: Context) -> None:
+    d = context.analysis_result.to_dict()
+    assert "top_files" in d, "Expected key 'top_files' in result dict"
+
+
+@then("the TierStats dict should have count {count:d} and size_bytes {size:d}")
+def step_assert_tier_stats_dict(context: Context, count: int, size: int) -> None:
+    d: dict[str, int] = context.result
+    assert d["count"] == count, f"Expected count {count}, got {d['count']}"
+    assert d["size_bytes"] == size, f"Expected size_bytes {size}, got {d['size_bytes']}"
+
+
+@then("the BudgetUtilization dict utilization_pct should be {expected:f}")
+def step_assert_budget_util_dict_pct(context: Context, expected: float) -> None:
+    d: dict[str, Any] = context.result
+    assert abs(d["utilization_pct"] - expected) < 0.01, (
+        f"Expected utilization_pct {expected}, got {d['utilization_pct']}"
+    )
+
+
+@then("the TopFileEntry dict should have all fields")
+def step_assert_top_file_dict(context: Context) -> None:
+    d: dict[str, Any] = context.result
+    for key in ("fragment_id", "resource_id", "access_count", "tier"):
+        assert key in d, f"Expected key {key!r} in TopFileEntry dict"
+
+
+@then("the max_bytes should be the hot-tier budget")
+def step_assert_max_bytes_hot_budget(context: Context) -> None:
+    util: BudgetUtilization = context.result
+    expected = context.tier_service.budget.max_tokens_hot
+    assert util.max_bytes == expected, (
+        f"Expected max_bytes {expected}, got {util.max_bytes}"
+    )
+
+
+@then('the CLI output should contain "ACMS Context Analysis"')
+def step_assert_cli_output_acms(context: Context) -> None:
+    assert "ACMS Context Analysis" in context.cli_output, (
+        f"Expected 'ACMS Context Analysis' in CLI output:\n{context.cli_output}"
+    )
+
+
+@then('the acms CLI JSON output should contain key "entry_count"')
+def step_assert_acms_cli_json_key(context: Context) -> None:
+    assert "entry_count" in context.acms_cli_json_data, (
+        "Expected key 'entry_count' in CLI JSON output"
+    )

src/cleveragents/application/services/context_analysis_engine.py

@@ -0,0 +1,328 @@
+"""Context Analysis Engine for ACMS index metrics.
+
+Provides actionable insight into the current ACMS state:
+
+| Metric              | Description                                                  |
+|---------------------|--------------------------------------------------------------|
+| ``entry_count``     | Total entries across all tiers                               |
+| ``tier_distribution``| Count and size per tier (hot/warm/cold)                     |
+| ``budget_utilization``| Current total size vs. configured max, as %               |
+| ``top_files``       | Top-N entries by access frequency (configurable N)           |
+
+The engine is wired to the ``ContextTierService`` and exposes both
+human-readable (text) and machine-readable (JSON) output via the
+``format_text`` and ``format_json`` helpers.
+
+Based on issue #9984 -- feat(acms): implement context analysis engine.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from cleveragents.application.services.context_tiers import ContextTierService
+from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment
+
+# ---------------------------------------------------------------------------
+# Result models (plain dataclasses -- no Pydantic to keep this lightweight)
+# ---------------------------------------------------------------------------
+
+
+class TierStats:
+    """Count and total size for a single tier."""
+
+    def __init__(self, count: int, size_bytes: int) -> None:
+        self.count = count
+        self.size_bytes = size_bytes
+
+    def to_dict(self) -> dict[str, int]:
+        return {"count": self.count, "size_bytes": self.size_bytes}
+
+
+class TierDistribution:
+    """Distribution of entries across hot/warm/cold tiers."""
+
+    def __init__(
+        self,
+        hot: TierStats,
+        warm: TierStats,
+        cold: TierStats,
+    ) -> None:
+        self.hot = hot
+        self.warm = warm
+        self.cold = cold
+
+    def to_dict(self) -> dict[str, dict[str, int]]:
+        return {
+            "hot": self.hot.to_dict(),
+            "warm": self.warm.to_dict(),
+            "cold": self.cold.to_dict(),
+        }
+
+
+class BudgetUtilization:
+    """Budget utilization metrics."""
+
+    def __init__(
+        self,
+        current_bytes: int,
+        max_bytes: int,
+        utilization_pct: float,
+    ) -> None:
+        self.current_bytes = current_bytes
+        self.max_bytes = max_bytes
+        self.utilization_pct = utilization_pct
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "current_bytes": self.current_bytes,
+            "max_bytes": self.max_bytes,
+            "utilization_pct": round(self.utilization_pct, 2),
+        }
+
+
+class TopFileEntry:
+    """A single entry in the top-files list."""
+
+    def __init__(
+        self,
+        fragment_id: str,
+        resource_id: str,
+        access_count: int,
+        tier: str,
+    ) -> None:
+        self.fragment_id = fragment_id
+        self.resource_id = resource_id
+        self.access_count = access_count
+        self.tier = tier
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "fragment_id": self.fragment_id,
+            "resource_id": self.resource_id,
+            "access_count": self.access_count,
+            "tier": self.tier,
+        }
+
+
+class AnalysisResult:
+    """Full analysis result from the ContextAnalysisEngine."""
+
+    def __init__(
+        self,
+        entry_count: int,
+        tier_distribution: TierDistribution,
+        budget_utilization: BudgetUtilization,
+        top_files: list[TopFileEntry],
+    ) -> None:
+        self.entry_count = entry_count
+        self.tier_distribution = tier_distribution
+        self.budget_utilization = budget_utilization
+        self.top_files = top_files
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "entry_count": self.entry_count,
+            "tier_distribution": self.tier_distribution.to_dict(),
+            "budget_utilization": self.budget_utilization.to_dict(),
+            "top_files": [f.to_dict() for f in self.top_files],
+        }
+
+
+# ---------------------------------------------------------------------------
+# Engine
+# ---------------------------------------------------------------------------
+
+
+class ContextAnalysisEngine:
+    """Query the ACMS index and produce analysis metrics.
+
+    Args:
+        tier_service: The ``ContextTierService`` to query.  When ``None``
+            a fresh in-memory service is created (useful for testing).
+        max_total_size: The configured maximum total size in bytes used
+            for budget utilisation calculation.  Defaults to the hot-tier
+            token budget from the service's ``TierBudget`` (treated as
+            bytes for simplicity when no explicit override is given).
+    """
+
+    def __init__(
+        self,
+        tier_service: ContextTierService | None = None,
+        max_total_size: int | None = None,
+    ) -> None:
+        self._tier_service: ContextTierService = (
+            tier_service if tier_service is not None else ContextTierService()
+        )
+        # Use explicit override or fall back to hot-tier token budget as proxy.
+        if max_total_size is not None:
+            self._max_total_size = max_total_size
+        else:
+            self._max_total_size = self._tier_service.budget.max_tokens_hot
+
+    # ------------------------------------------------------------------
+    # Individual metrics
+    # ------------------------------------------------------------------
+
+    def entry_count(self) -> int:
+        """Return the total number of entries across all tiers."""
+        metrics = self._tier_service.get_metrics()
+        return metrics.total_fragments
+
+    def tier_distribution(self) -> TierDistribution:
+        """Return count and total content size per tier.
+
+        Size is measured in bytes (``len(fragment.content.encode())``).
+        """
+        all_frags = self._tier_service.get_all_fragments()
+
+        hot_count = 0
+        hot_size = 0
+        warm_count = 0
+        warm_size = 0
+        cold_count = 0
+        cold_size = 0
+
+        for frag in all_frags:
+            size = len(frag.content.encode())
+            if frag.tier == ContextTier.HOT:
+                hot_count += 1
+                hot_size += size
+            elif frag.tier == ContextTier.WARM:
+                warm_count += 1
+                warm_size += size
+            else:
+                cold_count += 1
+                cold_size += size
+
+        return TierDistribution(
+            hot=TierStats(count=hot_count, size_bytes=hot_size),
+            warm=TierStats(count=warm_count, size_bytes=warm_size),
+            cold=TierStats(count=cold_count, size_bytes=cold_size),
+        )
+
+    def budget_utilization(self) -> BudgetUtilization:
+        """Return budget utilisation metrics.
+
+        ``current_bytes`` is the sum of encoded content sizes across all
+        tiers.  ``max_bytes`` is the configured ``max_total_size``.
+        ``utilization_pct`` is ``current_bytes / max_bytes * 100``,
+        capped at 100.0 when over budget.
+        """
+        all_frags = self._tier_service.get_all_fragments()
+        current_bytes = sum(len(f.content.encode()) for f in all_frags)
+        max_bytes = self._max_total_size
+
+        pct = min(current_bytes / max_bytes * 100.0, 100.0) if max_bytes > 0 else 0.0
+
+        return BudgetUtilization(
+            current_bytes=current_bytes,
+            max_bytes=max_bytes,
+            utilization_pct=pct,
+        )
+
+    def top_files(self, n: int = 10) -> list[TopFileEntry]:
+        """Return the top-N entries by ``access_count`` descending.
+
+        Args:
+            n: Number of entries to return (default 10).
+
+        Raises:
+            ValueError: If *n* is not positive.
+        """
+        if n < 1:
+            raise ValueError(f"n must be positive, got {n}")
+
+        all_frags: list[TieredFragment] = self._tier_service.get_all_fragments()
+        sorted_frags = sorted(
+            all_frags,
+            key=lambda f: f.access_count,
+            reverse=True,
+        )
+        return [
+            TopFileEntry(
+                fragment_id=frag.fragment_id,
+                resource_id=frag.resource_id,
+                access_count=frag.access_count,
+                tier=frag.tier.value,
+            )
+            for frag in sorted_frags[:n]
+        ]
+
+    # ------------------------------------------------------------------
+    # Full analysis
+    # ------------------------------------------------------------------
+
+    def analyze(self, top_n: int = 10) -> AnalysisResult:
+        """Run all metrics and return a combined ``AnalysisResult``.
+
+        Args:
+            top_n: Number of top files to include (default 10).
+        """
+        return AnalysisResult(
+            entry_count=self.entry_count(),
+            tier_distribution=self.tier_distribution(),
+            budget_utilization=self.budget_utilization(),
+            top_files=self.top_files(n=top_n),
+        )
+
+    # ------------------------------------------------------------------
+    # Formatters
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def format_json(result: AnalysisResult) -> str:
+        """Return a JSON string representation of *result*."""
+        return json.dumps(result.to_dict(), indent=2)
+
+    @staticmethod
+    def format_text(result: AnalysisResult) -> str:
+        """Return a human-readable text representation of *result*."""
+        lines: list[str] = []
+        lines.append("=== ACMS Context Analysis ===")
+        lines.append(f"Total entries: {result.entry_count}")
+        lines.append("")
+
+        dist = result.tier_distribution
+        lines.append("Tier Distribution:")
+        lines.append(
+            f"  hot:  {dist.hot.count:>6} entries, {dist.hot.size_bytes:>10} bytes"
+        )
+        lines.append(
+            f"  warm: {dist.warm.count:>6} entries, {dist.warm.size_bytes:>10} bytes"
+        )
+        lines.append(
+            f"  cold: {dist.cold.count:>6} entries, {dist.cold.size_bytes:>10} bytes"
+        )
+        lines.append("")
+
+        util = result.budget_utilization
+        lines.append("Budget Utilization:")
+        lines.append(f"  current: {util.current_bytes:>10} bytes")
+        lines.append(f"  max:     {util.max_bytes:>10} bytes")
+        lines.append(f"  used:    {util.utilization_pct:>9.2f}%")
+        lines.append("")
+
+        lines.append(f"Top {len(result.top_files)} Files by Access Frequency:")
+        if result.top_files:
+            for i, entry in enumerate(result.top_files, start=1):
+                resource = entry.resource_id or entry.fragment_id
+                lines.append(
+                    f"  {i:>3}. [{entry.tier:>4}] {resource}"
+                    f" (access_count={entry.access_count})"
+                )
+        else:
+            lines.append("  (no entries)")
+
+        return "\n".join(lines)
+
+
+__all__ = [
+    "AnalysisResult",
+    "BudgetUtilization",
+    "ContextAnalysisEngine",
+    "TierDistribution",
+    "TierStats",
+    "TopFileEntry",
+]

src/cleveragents/cli/commands/context.py

@@ -821,6 +821,57 @@ def context_clear(
         raise typer.Abort() from e
 
 
+@app.command("analyze")
+def context_analyze(
+    top_n: Annotated[
+        int,
+        typer.Option(
+            "--top",
+            "-n",
+            help="Number of top files to show by access frequency",
+        ),
+    ] = 10,
+    fmt: Annotated[
+        str,
+        typer.Option(
+            "--format",
+            "-f",
+            help="Output format: text (default) or json",
+        ),
+    ] = "text",
+    max_total_size: Annotated[
+        int | None,
+        typer.Option(
+            "--max-total-size",
+            help="Override configured max total size in bytes for budget utilization",
+        ),
+    ] = None,
+) -> None:
+    """Analyze the current ACMS context index.
+
+    Produces metrics including total entry count, tier distribution
+    (hot/warm/cold), budget utilization, and top files by access frequency.
+
+    Use ``--format json`` for machine-readable output.
+    """
+    from cleveragents.application.services.context_analysis_engine import (
+        ContextAnalysisEngine,
+    )
+    from cleveragents.application.services.context_tiers import ContextTierService
+
+    tier_service = ContextTierService()
+    engine = ContextAnalysisEngine(
+        tier_service=tier_service,
+        max_total_size=max_total_size,
+    )
+    result = engine.analyze(top_n=top_n)
+
+    if fmt.lower() == "json":
+        typer.echo(ContextAnalysisEngine.format_json(result))
+    else:
+        typer.echo(ContextAnalysisEngine.format_text(result))
+
+
 # Add a callback for when context command is called without subcommand
 @app.callback(invoke_without_command=True)
 def context_default(ctx: typer.Context) -> None: