From 3b66bf08f587144ee64957d2c0968298adf87d1a Mon Sep 17 00:00:00 2001 From: CleverThis Date: Sun, 19 Apr 2026 13:56:43 +0000 Subject: [PATCH] feat(acms): implement context analysis engine with tier distribution and budget utilization metrics Introduces a new ContextAnalysisEngine service to unify ACMS context analysis across tiered storage. It computes and exposes key observability metrics and provides flexible formatting for tooling and dashboards. Metrics include: - entry_count(): total entries across all tiers - tier_distribution(): per-tier counts and sizes (hot, warm, cold) - budget_utilization(): current total size vs. configured max, as a percentage - top_files(n): top-N entries by access_count in descending order - analyze(top_n): combined analysis result aggregating the above metrics - format_json() and format_text() static formatters for machine- and human-friendly output A new CLI command context analyze has been added at src/cleveragents/cli/commands/context.py to surface the feature from the command line. Unit tests live under features/acms_context_analysis_engine.feature (29 scenarios) with step definitions in features/steps/acms_context_analysis_engine_steps.py. The commit wires the new command into the CLI, enables test-driven validation of the metrics, and lays the groundwork for integration with dashboards. ISSUES CLOSED: #9984 --- features/acms_context_analysis_engine.feature | 195 ++++++ .../acms_context_analysis_engine_steps.py | 592 ++++++++++++++++++ .../services/context_analysis_engine.py | 328 ++++++++++ src/cleveragents/cli/commands/context.py | 51 ++ 4 files changed, 1166 insertions(+) create mode 100644 features/acms_context_analysis_engine.feature create mode 100644 features/steps/acms_context_analysis_engine_steps.py create mode 100644 src/cleveragents/application/services/context_analysis_engine.py diff --git a/features/acms_context_analysis_engine.feature b/features/acms_context_analysis_engine.feature new file mode 100644 index 000000000..a987d5a66 --- /dev/null +++ b/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" diff --git a/features/steps/acms_context_analysis_engine_steps.py b/features/steps/acms_context_analysis_engine_steps.py new file mode 100644 index 000000000..cee6fd835 --- /dev/null +++ b/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" + ) diff --git a/src/cleveragents/application/services/context_analysis_engine.py b/src/cleveragents/application/services/context_analysis_engine.py new file mode 100644 index 000000000..d79ff3fcd --- /dev/null +++ b/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", +] diff --git a/src/cleveragents/cli/commands/context.py b/src/cleveragents/cli/commands/context.py index 1e4a902bd..50959c1d9 100644 --- a/src/cleveragents/cli/commands/context.py +++ b/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: