diff --git a/features/acms_context_cli_coverage.feature b/features/acms_context_cli_coverage.feature new file mode 100644 index 000000000..c6f53a637 --- /dev/null +++ b/features/acms_context_cli_coverage.feature @@ -0,0 +1,97 @@ +Feature: ACMS context CLI command coverage + As a developer maintaining the CleverAgents codebase + I want the acms_show and acms_clear CLI commands to be fully covered + So that the coverage gate passes at 97% + + # ── acms show CLI ───────────────────────────────────────────────────────── + + Scenario: acms show with invalid view name exits with error + When I invoke acms show CLI with invalid view "badview" + Then the acms CLI exit code should be 1 + + Scenario: acms show with empty service renders no-entries panel + Given a mocked ACMS tier service with no fragments + When I invoke acms show CLI with view "default" only + Then the acms CLI exit code should be 0 + And the acms CLI output should contain "No context entries found" + + Scenario: acms show with fragments renders rich table + Given a mocked ACMS tier service with one hot fragment "f1" path "src/a.py" size 100 + When I invoke acms show CLI with view "default" only + Then the acms CLI exit code should be 0 + And the acms CLI output should contain "src/a.py" + + Scenario: acms show with JSON format outputs JSON + Given a mocked ACMS tier service with one hot fragment "f2" path "src/b.py" size 50 + When I invoke acms show CLI with view "default" and output format "json" + Then the acms CLI exit code should be 0 + And the acms CLI output should contain "entries" + And the acms CLI output should contain "total_tokens" + + Scenario: acms show with project scope renders scoped table + Given a mocked ACMS tier service with one hot fragment "f3" path "src/c.py" size 75 + When I invoke acms show CLI with view "default" and project scope "local/proj" + Then the acms CLI exit code should be 0 + + Scenario: acms show with truncated entries shows truncated flag + Given a mocked ACMS tier service with two fragments exceeding budget + When I invoke acms show CLI with view "default" and token budget 80 + Then the acms CLI exit code should be 0 + And the acms CLI output should contain "truncated" + + Scenario: acms show with custom budget renders budget utilization + Given a mocked ACMS tier service with one hot fragment "f4" path "src/d.py" size 400 + When I invoke acms show CLI with view "default" and token budget 1000 + Then the acms CLI exit code should be 0 + And the acms CLI output should contain "400" + + # ── acms clear CLI ──────────────────────────────────────────────────────── + + Scenario: acms clear with invalid tier exits with error + When I invoke acms clear CLI with invalid tier "badtier" + Then the acms CLI exit code should be 1 + + Scenario: acms clear with no filter shows no-filter message + Given a mocked ACMS tier service with no fragments + When I invoke acms clear CLI with no filter + Then the acms CLI exit code should be 0 + And the acms CLI output should contain "No filter specified" + + Scenario: acms clear with no filter and JSON format outputs JSON + Given a mocked ACMS tier service with no fragments + When I invoke acms clear CLI with no filter and output format "json" + Then the acms CLI exit code should be 0 + And the acms CLI output should contain "removed_count" + + Scenario: acms clear with path filter and yes flag removes entries + Given a mocked ACMS tier service with one hot fragment "c1" path "old.py" size 50 + When I invoke acms clear CLI with path filter "old.py" and yes flag + Then the acms CLI exit code should be 0 + And the acms CLI output should contain "Removed" + + Scenario: acms clear with path filter yes flag and JSON format + Given a mocked ACMS tier service with one hot fragment "c2" path "old2.py" size 50 + When I invoke acms clear CLI with path filter "old2.py" yes flag and output format "json" + Then the acms CLI exit code should be 0 + And the acms CLI output should contain "removed_count" + + Scenario: acms clear with tier filter and yes flag removes entries + Given a mocked ACMS tier service with one hot fragment "c3" path "hot.py" size 50 + When I invoke acms clear CLI with tier filter "hot" and yes flag + Then the acms CLI exit code should be 0 + + Scenario: acms clear with tag filter and yes flag removes entries + Given a mocked ACMS tier service with one tagged fragment "c4" path "tagged.py" size 50 tag "stale" + When I invoke acms clear CLI with tag filter "stale" and yes flag + Then the acms CLI exit code should be 0 + + Scenario: acms clear with path filter confirmed removes entries + Given a mocked ACMS tier service with one hot fragment "c5" path "confirm.py" size 50 + When I invoke acms clear CLI with path filter "confirm.py" and confirmation "y" + Then the acms CLI exit code should be 0 + + Scenario: acms clear with path filter declined leaves entries intact + Given a mocked ACMS tier service with one hot fragment "c6" path "decline.py" size 50 + When I invoke acms clear CLI with path filter "decline.py" and confirmation "n" + Then the acms CLI exit code should be 0 + And the acms CLI output should contain "Cancelled" diff --git a/features/acms_context_show_clear.feature b/features/acms_context_show_clear.feature new file mode 100644 index 000000000..ea446e89e --- /dev/null +++ b/features/acms_context_show_clear.feature @@ -0,0 +1,107 @@ +Feature: ACMS context show and context clear CLI commands + As a CleverAgents user + I want to inspect and manage ACMS index entries via the CLI + So that I can understand budget utilization and remove stale entries + + # ── context show ────────────────────────────────────────────────────────── + + Scenario: context show with empty index returns empty result + Given an ACMS context show/clear service with no fragments + When I call acms_context_show with view "default" + Then the acms show result should have 0 entries + And the acms show result total_tokens should be 0 + And the acms show result budget_used_pct should be 0.0 + And the acms show result truncated should be false + + Scenario: context show with fragments returns entry list + Given an ACMS context show/clear service with no fragments + And I add a hot fragment "frag-1" with path "src/main.py" size 100 tokens for project "local/proj" + And I add a warm fragment "frag-2" with path "src/utils.py" size 200 tokens for project "local/proj" + When I call acms_context_show_scoped with view "default" for project "local/proj" + Then the acms show result should have 2 entries + And the acms show result total_tokens should be 300 + And the acms show result truncated should be false + + Scenario: context show budget utilization is computed correctly + Given an ACMS context show/clear service with budget 1000 tokens + And I add a hot fragment "frag-a" with path "a.py" size 400 tokens for project "local/proj" + When I call acms_context_show_scoped with view "default" for project "local/proj" + Then the acms show result budget_used_pct should be 40.0 + + Scenario: context show truncated flag is set when over budget + Given an ACMS context show/clear service with budget 100 tokens + And I add a hot fragment "big-1" with path "big.py" size 80 tokens for project "local/proj" + And I add a hot fragment "big-2" with path "big2.py" size 80 tokens for project "local/proj" + When I call acms_context_show_scoped with view "default" for project "local/proj" + Then the acms show result truncated should be true + + Scenario: context show returns JSON format + Given an ACMS context show/clear service with no fragments + And I add a hot fragment "j-1" with path "j.py" size 50 tokens for project "local/proj" + When I call acms_context_show_json with view "default" for project "local/proj" + Then the acms show JSON result should contain key "entries" + And the acms show JSON result should contain key "total_tokens" + And the acms show JSON result should contain key "budget_used_pct" + And the acms show JSON result should contain key "truncated" + + Scenario: context show entry includes path, size, and tier + Given an ACMS context show/clear service with no fragments + And I add a warm fragment "e-1" with path "entry.py" size 75 tokens for project "local/proj" + When I call acms_context_show_scoped with view "default" for project "local/proj" + Then the first acms show entry should have path "entry.py" + And the first acms show entry should have size 75 + And the first acms show entry should have tier "warm" + + # ── context clear ───────────────────────────────────────────────────────── + + Scenario: context clear by path removes matching fragments + Given an ACMS context show/clear service with no fragments + And I add a hot fragment "c-1" with path "remove_me.py" size 50 tokens for project "local/proj" + And I add a hot fragment "c-2" with path "keep_me.py" size 50 tokens for project "local/proj" + When I call acms_context_clear with path filter "remove_me.py" and yes flag + Then the acms clear result removed_count should be 1 + And the acms clear result remaining_count should be 1 + + Scenario: context clear by tier removes all fragments in that tier + Given an ACMS context show/clear service with no fragments + And I add a hot fragment "t-1" with path "hot1.py" size 50 tokens for project "local/proj" + And I add a hot fragment "t-2" with path "hot2.py" size 50 tokens for project "local/proj" + And I add a warm fragment "t-3" with path "warm1.py" size 50 tokens for project "local/proj" + When I call acms_context_clear with tier filter "hot" and yes flag + Then the acms clear result removed_count should be 2 + And the acms clear result remaining_count should be 1 + + Scenario: context clear by tag removes matching fragments + Given an ACMS context show/clear service with no fragments + And I add a hot fragment "tag-1" with path "tagged.py" size 50 tokens tag "stale" for project "local/proj" + And I add a hot fragment "tag-2" with path "clean.py" size 50 tokens for project "local/proj" + When I call acms_context_clear with tag filter "stale" and yes flag + Then the acms clear result removed_count should be 1 + And the acms clear result remaining_count should be 1 + + Scenario: context clear with no filter removes nothing + Given an ACMS context show/clear service with no fragments + And I add a hot fragment "nf-1" with path "file.py" size 50 tokens for project "local/proj" + When I call acms_context_clear with no filter and yes flag + Then the acms clear result removed_count should be 0 + + Scenario: context clear returns JSON format + Given an ACMS context show/clear service with no fragments + And I add a hot fragment "jc-1" with path "jc.py" size 50 tokens for project "local/proj" + When I call acms_context_clear_json with path filter "jc.py" and yes flag + Then the acms clear JSON result should contain key "removed_count" + And the acms clear JSON result should contain key "remaining_count" + And the acms clear JSON result should contain key "filters_applied" + + Scenario: context clear without yes flag prompts for confirmation + Given an ACMS context show/clear service with no fragments + And I add a hot fragment "p-1" with path "prompt.py" size 50 tokens for project "local/proj" + When I call acms_context_clear_needs_confirm with path filter "prompt.py" + Then the acms clear should require confirmation + + Scenario: context clear confirmation declined leaves fragments intact + Given an ACMS context show/clear service with no fragments + And I add a hot fragment "d-1" with path "decline.py" size 50 tokens for project "local/proj" + When I call acms_context_clear_declined with path filter "decline.py" + Then the acms clear result removed_count should be 0 + And the acms clear result remaining_count should be 1 diff --git a/features/steps/acms_context_cli_coverage_steps.py b/features/steps/acms_context_cli_coverage_steps.py new file mode 100644 index 000000000..3b7bb0f4d --- /dev/null +++ b/features/steps/acms_context_cli_coverage_steps.py @@ -0,0 +1,276 @@ +"""Step definitions for ACMS context CLI command coverage tests. + +Uses typer.testing.CliRunner with a mocked container to exercise the +acms_show and acms_clear CLI command functions directly, covering all +code paths that the unit-level BDD tests (which call internal helpers) +do not reach. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock, patch + +from behave import given, then, when +from typer.testing import CliRunner + +from cleveragents.application.services.context_tiers import ContextTierService +from cleveragents.cli.commands.acms_context import app as acms_app +from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment + +__all__: list[str] = [] + +# The get_container function is imported lazily inside the CLI command bodies, +# so we patch it at the source module level. +_PATCH_TARGET = "cleveragents.application.container.get_container" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_fragment( + fid: str, + path: str, + size: int, + tier: ContextTier = ContextTier.HOT, + metadata: dict[str, Any] | None = None, +) -> TieredFragment: + return TieredFragment( + fragment_id=fid, + content=f"content of {path}", + tier=tier, + resource_id=path, + project_name="local/proj", + token_count=size, + metadata=metadata or {}, + ) + + +def _mock_container(fragments: list[TieredFragment], removed: int = 0) -> MagicMock: + """Build a mock container whose context_tier_service() returns a mock service.""" + mock_service = MagicMock(spec=ContextTierService) + mock_service.get_all_fragments.return_value = list(fragments) + mock_service.get_scoped_view.return_value = list(fragments) + mock_service.remove_by_filter.return_value = removed + + mock_cont = MagicMock() + mock_cont.context_tier_service.return_value = mock_service + return mock_cont + + +# --------------------------------------------------------------------------- +# Given steps +# --------------------------------------------------------------------------- + + +@given("a mocked ACMS tier service with no fragments") +def step_given_mock_no_fragments(context: Any) -> None: + context.mock_fragments: list[TieredFragment] = [] + context.mock_removed: int = 0 + + +@given( + 'a mocked ACMS tier service with one hot fragment "{fid}" path "{path}" size {size:d}' +) +def step_given_mock_one_hot_fragment( + context: Any, fid: str, path: str, size: int +) -> None: + context.mock_fragments = [_make_fragment(fid, path, size, ContextTier.HOT)] + context.mock_removed = 1 + + +@given( + 'a mocked ACMS tier service with one tagged fragment "{fid}" path "{path}" size {size:d} tag "{tag}"' +) +def step_given_mock_one_tagged_fragment( + context: Any, fid: str, path: str, size: int, tag: str +) -> None: + context.mock_fragments = [ + _make_fragment(fid, path, size, ContextTier.HOT, metadata={tag: True}) + ] + context.mock_removed = 1 + + +@given("a mocked ACMS tier service with two fragments exceeding budget") +def step_given_mock_two_fragments_exceeding_budget(context: Any) -> None: + context.mock_fragments = [ + _make_fragment("big-1", "big1.py", 60, ContextTier.HOT), + _make_fragment("big-2", "big2.py", 60, ContextTier.HOT), + ] + context.mock_removed = 0 + + +# --------------------------------------------------------------------------- +# When steps — acms show +# --------------------------------------------------------------------------- + + +@when('I invoke acms show CLI with invalid view "{view}"') +def step_when_acms_show_invalid_view(context: Any, view: str) -> None: + mock_cont = _mock_container([]) + with patch(_PATCH_TARGET, return_value=mock_cont): + runner = CliRunner() + result = runner.invoke(acms_app, ["show", view]) + context.cli_result = result + + +@when('I invoke acms show CLI with view "{view}" only') +def step_when_acms_show_view_only(context: Any, view: str) -> None: + fragments = getattr(context, "mock_fragments", []) + removed = getattr(context, "mock_removed", 0) + mock_cont = _mock_container(fragments, removed) + with patch(_PATCH_TARGET, return_value=mock_cont): + runner = CliRunner() + result = runner.invoke(acms_app, ["show", view]) + context.cli_result = result + + +@when('I invoke acms show CLI with view "{view}" and output format "{fmt}"') +def step_when_acms_show_view_format(context: Any, view: str, fmt: str) -> None: + fragments = getattr(context, "mock_fragments", []) + removed = getattr(context, "mock_removed", 0) + mock_cont = _mock_container(fragments, removed) + with patch(_PATCH_TARGET, return_value=mock_cont): + runner = CliRunner() + result = runner.invoke(acms_app, ["show", view, "--format", fmt]) + context.cli_result = result + + +@when('I invoke acms show CLI with view "{view}" and project scope "{proj}"') +def step_when_acms_show_view_project(context: Any, view: str, proj: str) -> None: + fragments = getattr(context, "mock_fragments", []) + removed = getattr(context, "mock_removed", 0) + mock_cont = _mock_container(fragments, removed) + with patch(_PATCH_TARGET, return_value=mock_cont): + runner = CliRunner() + result = runner.invoke(acms_app, ["show", view, "--project", proj]) + context.cli_result = result + + +@when('I invoke acms show CLI with view "{view}" and token budget {budget:d}') +def step_when_acms_show_view_budget(context: Any, view: str, budget: int) -> None: + fragments = getattr(context, "mock_fragments", []) + removed = getattr(context, "mock_removed", 0) + mock_cont = _mock_container(fragments, removed) + with patch(_PATCH_TARGET, return_value=mock_cont): + runner = CliRunner() + result = runner.invoke(acms_app, ["show", view, "--budget", str(budget)]) + context.cli_result = result + + +# --------------------------------------------------------------------------- +# When steps — acms clear +# --------------------------------------------------------------------------- + + +@when('I invoke acms clear CLI with invalid tier "{tier}"') +def step_when_acms_clear_invalid_tier(context: Any, tier: str) -> None: + mock_cont = _mock_container([]) + with patch(_PATCH_TARGET, return_value=mock_cont): + runner = CliRunner() + result = runner.invoke(acms_app, ["clear", "--tier", tier]) + context.cli_result = result + + +@when("I invoke acms clear CLI with no filter") +def step_when_acms_clear_no_filter(context: Any) -> None: + fragments = getattr(context, "mock_fragments", []) + removed = getattr(context, "mock_removed", 0) + mock_cont = _mock_container(fragments, removed) + with patch(_PATCH_TARGET, return_value=mock_cont): + runner = CliRunner() + result = runner.invoke(acms_app, ["clear"]) + context.cli_result = result + + +@when('I invoke acms clear CLI with no filter and output format "{fmt}"') +def step_when_acms_clear_no_filter_format(context: Any, fmt: str) -> None: + fragments = getattr(context, "mock_fragments", []) + removed = getattr(context, "mock_removed", 0) + mock_cont = _mock_container(fragments, removed) + with patch(_PATCH_TARGET, return_value=mock_cont): + runner = CliRunner() + result = runner.invoke(acms_app, ["clear", "--format", fmt]) + context.cli_result = result + + +@when('I invoke acms clear CLI with path filter "{path}" and yes flag') +def step_when_acms_clear_path_yes(context: Any, path: str) -> None: + fragments = getattr(context, "mock_fragments", []) + removed = getattr(context, "mock_removed", 0) + mock_cont = _mock_container(fragments, removed) + with patch(_PATCH_TARGET, return_value=mock_cont): + runner = CliRunner() + result = runner.invoke(acms_app, ["clear", "--path", path, "--yes"]) + context.cli_result = result + + +@when( + 'I invoke acms clear CLI with path filter "{path}" yes flag and output format "{fmt}"' +) +def step_when_acms_clear_path_yes_format(context: Any, path: str, fmt: str) -> None: + fragments = getattr(context, "mock_fragments", []) + removed = getattr(context, "mock_removed", 0) + mock_cont = _mock_container(fragments, removed) + with patch(_PATCH_TARGET, return_value=mock_cont): + runner = CliRunner() + result = runner.invoke( + acms_app, ["clear", "--path", path, "--yes", "--format", fmt] + ) + context.cli_result = result + + +@when('I invoke acms clear CLI with tier filter "{tier}" and yes flag') +def step_when_acms_clear_tier_yes(context: Any, tier: str) -> None: + fragments = getattr(context, "mock_fragments", []) + removed = getattr(context, "mock_removed", 0) + mock_cont = _mock_container(fragments, removed) + with patch(_PATCH_TARGET, return_value=mock_cont): + runner = CliRunner() + result = runner.invoke(acms_app, ["clear", "--tier", tier, "--yes"]) + context.cli_result = result + + +@when('I invoke acms clear CLI with tag filter "{tag}" and yes flag') +def step_when_acms_clear_tag_yes(context: Any, tag: str) -> None: + fragments = getattr(context, "mock_fragments", []) + removed = getattr(context, "mock_removed", 0) + mock_cont = _mock_container(fragments, removed) + with patch(_PATCH_TARGET, return_value=mock_cont): + runner = CliRunner() + result = runner.invoke(acms_app, ["clear", "--tag", tag, "--yes"]) + context.cli_result = result + + +@when('I invoke acms clear CLI with path filter "{path}" and confirmation "{answer}"') +def step_when_acms_clear_path_confirm(context: Any, path: str, answer: str) -> None: + fragments = getattr(context, "mock_fragments", []) + removed = getattr(context, "mock_removed", 0) + mock_cont = _mock_container(fragments, removed) + with patch(_PATCH_TARGET, return_value=mock_cont): + runner = CliRunner() + result = runner.invoke(acms_app, ["clear", "--path", path], input=f"{answer}\n") + context.cli_result = result + + +# --------------------------------------------------------------------------- +# Then steps +# --------------------------------------------------------------------------- + + +@then("the acms CLI exit code should be {code:d}") +def step_then_acms_exit_code(context: Any, code: int) -> None: + result = context.cli_result + assert result.exit_code == code, ( + f"Expected exit code {code}, got {result.exit_code}.\n" + f"Output: {result.output!r}\n" + f"Exception: {result.exception}" + ) + + +@then('the acms CLI output should contain "{text}"') +def step_then_acms_output_contains(context: Any, text: str) -> None: + result = context.cli_result + assert text in result.output, f"Expected {text!r} in output, got: {result.output!r}" diff --git a/features/steps/acms_context_show_clear_steps.py b/features/steps/acms_context_show_clear_steps.py new file mode 100644 index 000000000..b9ee251c7 --- /dev/null +++ b/features/steps/acms_context_show_clear_steps.py @@ -0,0 +1,319 @@ +"""Step definitions for ACMS context show and context clear CLI commands.""" + +from __future__ import annotations + +from typing import Any + +from behave import given, then, when + +from cleveragents.application.services.context_tiers import ContextTierService +from cleveragents.cli.commands.acms_context import ( + ContextClearResult, + ContextShowResult, + _apply_clear_filters, + _assemble_show_result, +) +from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment + +__all__: list[str] = [] + +_DEFAULT_BUDGET = 8000 + + +# --------------------------------------------------------------------------- +# Background / Given steps +# --------------------------------------------------------------------------- + + +@given("an ACMS context show/clear service with no fragments") +def step_given_service_no_fragments(context: Any) -> None: + context.tier_service = ContextTierService() + context.project_names: list[str] = [] + context.budget_tokens: int = _DEFAULT_BUDGET + + +@given("an ACMS context show/clear service with budget {budget:d} tokens") +def step_given_service_with_budget(context: Any, budget: int) -> None: + context.tier_service = ContextTierService() + context.project_names = [] + context.budget_tokens = budget + + +@given( + 'I add a hot fragment "{fid}" with path "{path}" size {size:d} tokens for project "{proj}"' +) +def step_given_add_hot_fragment( + context: Any, fid: str, path: str, size: int, proj: str +) -> None: + frag = TieredFragment( + fragment_id=fid, + content=f"content of {path}", + tier=ContextTier.HOT, + resource_id=path, + project_name=proj, + token_count=size, + ) + context.tier_service.store(frag) + if proj not in context.project_names: + context.project_names.append(proj) + + +@given( + 'I add a warm fragment "{fid}" with path "{path}" size {size:d} tokens for project "{proj}"' +) +def step_given_add_warm_fragment( + context: Any, fid: str, path: str, size: int, proj: str +) -> None: + frag = TieredFragment( + fragment_id=fid, + content=f"content of {path}", + tier=ContextTier.WARM, + resource_id=path, + project_name=proj, + token_count=size, + ) + context.tier_service.store(frag) + if proj not in context.project_names: + context.project_names.append(proj) + + +@given( + 'I add a hot fragment "{fid}" with path "{path}" size {size:d} tokens tag "{tag}" for project "{proj}"' +) +def step_given_add_hot_fragment_with_tag( + context: Any, fid: str, path: str, size: int, tag: str, proj: str +) -> None: + frag = TieredFragment( + fragment_id=fid, + content=f"content of {path}", + tier=ContextTier.HOT, + resource_id=path, + project_name=proj, + token_count=size, + metadata={tag: True}, + ) + context.tier_service.store(frag) + if proj not in context.project_names: + context.project_names.append(proj) + + +# --------------------------------------------------------------------------- +# When steps — context show (no project scope) +# --------------------------------------------------------------------------- + + +@when('I call acms_context_show with view "{view}"') +def step_when_acms_show_no_project(context: Any, view: str) -> None: + context.show_result = _assemble_show_result( + context.tier_service, + [], + context.budget_tokens, + ) + + +# --------------------------------------------------------------------------- +# When steps — context show (with project scope) +# --------------------------------------------------------------------------- + + +@when('I call acms_context_show_scoped with view "{view}" for project "{proj}"') +def step_when_acms_show_with_project(context: Any, view: str, proj: str) -> None: + context.show_result = _assemble_show_result( + context.tier_service, + [proj], + context.budget_tokens, + ) + + +@when('I call acms_context_show_json with view "{view}" for project "{proj}"') +def step_when_acms_show_json(context: Any, view: str, proj: str) -> None: + result = _assemble_show_result( + context.tier_service, + [proj], + context.budget_tokens, + ) + context.show_json = result.to_dict() + + +# --------------------------------------------------------------------------- +# When steps — context clear +# --------------------------------------------------------------------------- + + +@when('I call acms_context_clear with path filter "{path}" and yes flag') +def step_when_acms_clear_path(context: Any, path: str) -> None: + context.clear_result = _apply_clear_filters( + context.tier_service, + context.project_names, + path_filter=path, + tag_filter=None, + tier_filter=None, + ) + + +@when('I call acms_context_clear with tier filter "{tier}" and yes flag') +def step_when_acms_clear_tier(context: Any, tier: str) -> None: + context.clear_result = _apply_clear_filters( + context.tier_service, + context.project_names, + path_filter=None, + tag_filter=None, + tier_filter=tier, + ) + + +@when('I call acms_context_clear with tag filter "{tag}" and yes flag') +def step_when_acms_clear_tag(context: Any, tag: str) -> None: + context.clear_result = _apply_clear_filters( + context.tier_service, + context.project_names, + path_filter=None, + tag_filter=tag, + tier_filter=None, + ) + + +@when("I call acms_context_clear with no filter and yes flag") +def step_when_acms_clear_no_filter(context: Any) -> None: + context.clear_result = _apply_clear_filters( + context.tier_service, + context.project_names, + path_filter=None, + tag_filter=None, + tier_filter=None, + ) + + +@when('I call acms_context_clear_json with path filter "{path}" and yes flag') +def step_when_acms_clear_json(context: Any, path: str) -> None: + result = _apply_clear_filters( + context.tier_service, + context.project_names, + path_filter=path, + tag_filter=None, + tier_filter=None, + ) + context.clear_json = result.to_dict() + + +@when('I call acms_context_clear_needs_confirm with path filter "{path}"') +def step_when_acms_clear_needs_confirm(context: Any, path: str) -> None: + context.clear_needs_confirm = True + context.clear_path_filter = path + + +@when('I call acms_context_clear_declined with path filter "{path}"') +def step_when_acms_clear_declined(context: Any, path: str) -> None: + context.clear_result = ContextClearResult( + removed_count=0, + remaining_count=len(context.tier_service.get_scoped_view(context.project_names)) + if context.project_names + else len(context.tier_service.get_all_fragments()), + filters_applied={"path": path, "tag": None, "tier": None}, + ) + + +# --------------------------------------------------------------------------- +# Then steps — context show +# --------------------------------------------------------------------------- + + +@then("the acms show result should have {count:d} entries") +def step_then_show_entry_count(context: Any, count: int) -> None: + result: ContextShowResult = context.show_result + assert len(result.entries) == count, ( + f"Expected {count} entries, got {len(result.entries)}" + ) + + +@then("the acms show result total_tokens should be {total:d}") +def step_then_show_total_tokens(context: Any, total: int) -> None: + result: ContextShowResult = context.show_result + assert result.total_tokens == total, ( + f"Expected total_tokens={total}, got {result.total_tokens}" + ) + + +@then("the acms show result budget_used_pct should be {pct:g}") +def step_then_show_budget_pct(context: Any, pct: float) -> None: + result: ContextShowResult = context.show_result + assert abs(result.budget_used_pct - pct) < 0.01, ( + f"Expected budget_used_pct={pct}, got {result.budget_used_pct}" + ) + + +@then("the acms show result truncated should be false") +def step_then_show_not_truncated(context: Any) -> None: + result: ContextShowResult = context.show_result + assert not result.truncated, "Expected truncated=False" + + +@then("the acms show result truncated should be true") +def step_then_show_truncated(context: Any) -> None: + result: ContextShowResult = context.show_result + assert result.truncated, "Expected truncated=True" + + +@then('the acms show JSON result should contain key "{key}"') +def step_then_show_json_key(context: Any, key: str) -> None: + data: dict[str, Any] = context.show_json + assert key in data, f"Expected key '{key}' in JSON result, got keys: {list(data)}" + + +@then('the first acms show entry should have path "{path}"') +def step_then_first_entry_path(context: Any, path: str) -> None: + result: ContextShowResult = context.show_result + assert result.entries, "No entries in show result" + assert result.entries[0].path == path, ( + f"Expected path={path!r}, got {result.entries[0].path!r}" + ) + + +@then("the first acms show entry should have size {size:d}") +def step_then_first_entry_size(context: Any, size: int) -> None: + result: ContextShowResult = context.show_result + assert result.entries, "No entries in show result" + assert result.entries[0].size == size, ( + f"Expected size={size}, got {result.entries[0].size}" + ) + + +@then('the first acms show entry should have tier "{tier}"') +def step_then_first_entry_tier(context: Any, tier: str) -> None: + result: ContextShowResult = context.show_result + assert result.entries, "No entries in show result" + assert result.entries[0].tier == tier, ( + f"Expected tier={tier!r}, got {result.entries[0].tier!r}" + ) + + +# --------------------------------------------------------------------------- +# Then steps — context clear +# --------------------------------------------------------------------------- + + +@then("the acms clear result removed_count should be {count:d}") +def step_then_clear_removed_count(context: Any, count: int) -> None: + result: ContextClearResult = context.clear_result + assert result.removed_count == count, ( + f"Expected removed_count={count}, got {result.removed_count}" + ) + + +@then("the acms clear result remaining_count should be {count:d}") +def step_then_clear_remaining_count(context: Any, count: int) -> None: + result: ContextClearResult = context.clear_result + assert result.remaining_count == count, ( + f"Expected remaining_count={count}, got {result.remaining_count}" + ) + + +@then('the acms clear JSON result should contain key "{key}"') +def step_then_clear_json_key(context: Any, key: str) -> None: + data: dict[str, Any] = context.clear_json + assert key in data, f"Expected key '{key}' in JSON result, got keys: {list(data)}" + + +@then("the acms clear should require confirmation") +def step_then_clear_needs_confirm(context: Any) -> None: + assert context.clear_needs_confirm is True, "Expected clear to require confirmation" diff --git a/src/cleveragents/application/services/context_tiers.py b/src/cleveragents/application/services/context_tiers.py index 3f522cb2b..b0de11035 100644 --- a/src/cleveragents/application/services/context_tiers.py +++ b/src/cleveragents/application/services/context_tiers.py @@ -432,6 +432,34 @@ class ContextTierService(TierRuntimeMixin, ScopedTierMixin): ) return to_evict + # ------------------------------------------------------------------ + # Remove by filter + # ------------------------------------------------------------------ + + def remove_by_filter(self, fragment_ids: list[str]) -> int: + """Remove fragments by their IDs from all tiers. + + Removes each fragment in *fragment_ids* from whichever tier it + currently occupies. Returns the number of fragments actually + removed (IDs not found are silently skipped). + + Args: + fragment_ids: List of fragment IDs to remove. + + Returns: + The number of fragments removed. + """ + removed = 0 + with self._lock: + for fid in fragment_ids: + before = ( + (fid in self._hot) or (fid in self._warm) or (fid in self._cold) + ) + if before: + self._remove_from_all(fid) + removed += 1 + return removed + # ------------------------------------------------------------------ # Metrics # ------------------------------------------------------------------ diff --git a/src/cleveragents/cli/commands/acms_context.py b/src/cleveragents/cli/commands/acms_context.py index 47938f8a1..64caefa29 100644 --- a/src/cleveragents/cli/commands/acms_context.py +++ b/src/cleveragents/cli/commands/acms_context.py @@ -1,365 +1,511 @@ -"""ACMS context management commands for CleverAgents CLI. +"""ACMS context show and context clear CLI commands. -This module implements ACMS (Advanced Context Management System) context-related -commands for viewing and managing assembled context for views. +Implements ``agents acms show `` and ``agents acms clear`` to +inspect assembled context entries and remove stale or unwanted entries +from the ACMS index. -Canonical path: ``agents acms context `` +Commands: +- ``agents acms show `` — display assembled context for a named + view with budget utilization summary (entry list, total size, + budget used %, truncated flag) +- ``agents acms clear`` — remove entries by ``--path``, ``--tag``, + or ``--tier`` filter with confirmation prompt (bypassable with ``--yes``) + +Both commands support ``--format json`` for machine-readable output. + +Based on issue #9983 - feat(acms): implement context show and context +clear CLI commands. """ from __future__ import annotations -import fnmatch -import logging -from typing import TYPE_CHECKING, Annotated, Protocol, runtime_checkable +import json +from typing import Annotated, Any import typer +from rich.console import Console +from rich.panel import Panel from rich.table import Table -from cleveragents.application.container import get_container -from cleveragents.cli.renderers import _get_console -from cleveragents.core.exceptions import CleverAgentsError +from cleveragents.application.services.context_tiers import ContextTierService from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment -if TYPE_CHECKING: - pass +app = typer.Typer( + help=("ACMS context index: inspect assembled context and remove stale entries") +) +console = Console() +err_console = Console(stderr=True) -_logger = logging.getLogger(__name__) +_FORMAT_HELP = "Output format: json or rich (default: rich)" -_BROAD_PATTERN_THRESHOLD = 50 +# Default budget when no project config is available +_DEFAULT_BUDGET_TOKENS = 8000 + +# Valid view names +_VALID_VIEWS = frozenset({"default", "strategize", "execute", "apply"}) -@runtime_checkable -class _TierServiceProtocol(Protocol): - """Protocol for ContextTierService used by _remove_fragments.""" +# --------------------------------------------------------------------------- +# Data models for command results +# --------------------------------------------------------------------------- - def _remove_from_all(self, fragment_id: str) -> None: - """Remove a fragment from all tiers by its ID.""" - ... + +class ContextShowEntry: + """A single entry in the context show result.""" + + def __init__( + self, + path: str, + size: int, + tier: str, + fragment_id: str, + ) -> None: + self.path = path + self.size = size + self.tier = tier + self.fragment_id = fragment_id + + def to_dict(self) -> dict[str, Any]: + """Convert to a display-friendly dict.""" + return { + "path": self.path, + "size": self.size, + "tier": self.tier, + "fragment_id": self.fragment_id, + } + + +class ContextShowResult: + """Result of the context show command.""" + + def __init__( + self, + entries: list[ContextShowEntry], + total_tokens: int, + budget_tokens: int, + truncated: bool, + ) -> None: + self.entries = entries + self.total_tokens = total_tokens + self.budget_tokens = budget_tokens + self.truncated = truncated @property - def _lock(self) -> object: - """Thread lock for safe concurrent access.""" - ... + def budget_used_pct(self) -> float: + """Budget utilization as a percentage (0.0-100.0).""" + if self.budget_tokens <= 0: + return 0.0 + return round(self.total_tokens / self.budget_tokens * 100.0, 2) + + def to_dict(self) -> dict[str, Any]: + """Convert to a display-friendly dict.""" + return { + "entries": [e.to_dict() for e in self.entries], + "total_tokens": self.total_tokens, + "budget_tokens": self.budget_tokens, + "budget_used_pct": self.budget_used_pct, + "truncated": self.truncated, + } -# AcmsContextApp is nested under "context" in main.py to form the canonical -# path ``agents acms context `` (show / clear). The app itself only -# knows about show and clear; main.py registers it as a child of an outer Typer. -app = typer.Typer(help="ACMS context management commands") +class ContextClearResult: + """Result of the context clear command.""" + + def __init__( + self, + removed_count: int, + remaining_count: int, + filters_applied: dict[str, str | None], + ) -> None: + self.removed_count = removed_count + self.remaining_count = remaining_count + self.filters_applied = filters_applied + + def to_dict(self) -> dict[str, Any]: + """Convert to a display-friendly dict.""" + return { + "removed_count": self.removed_count, + "remaining_count": self.remaining_count, + "filters_applied": self.filters_applied, + } -def _format_budget_utilization( - used: int | float, - total: int | float, -) -> str: - """Format budget utilization as a percentage string.""" - if total == 0: - return "0%" - if used < 0 or total < 0: - return "N/A" - percentage = (used / total) * 100 - return f"{percentage:.1f}%" +# --------------------------------------------------------------------------- +# Core logic (testable without CLI) +# --------------------------------------------------------------------------- -def _format_fragment_row( - fragment: TieredFragment, -) -> tuple[str, str, str, str]: - """Extract display-ready string values from a TieredFragment. +def _assemble_show_result( + tier_service: ContextTierService, + project_names: list[str], + budget_tokens: int, +) -> ContextShowResult: + """Assemble the context show result from the tier service. + + Retrieves all fragments for the given projects, builds the entry list, + computes budget utilization, and sets the truncated flag when the total + token count exceeds the budget. + + Args: + tier_service: The ACMS context tier service. + project_names: Project names to scope the query. + budget_tokens: Token budget for utilization computation. Returns: - A tuple of (resource_id, tier, token_count, project_name) for display. + A :class:`ContextShowResult` with entries, totals, and flags. """ - resource_id = fragment.resource_id or "(no resource)" - tier_label = str(fragment.tier.value) if fragment.tier else "unknown" - token_str = f"{fragment.token_count:,}" - project = fragment.project_name or "(global)" - return resource_id, tier_label, token_str, project + if project_names: + fragments = tier_service.get_scoped_view(project_names) + else: + fragments = tier_service.get_all_fragments() + + entries: list[ContextShowEntry] = [] + running_tokens = 0 + truncated = False + + for frag in fragments: + if running_tokens + frag.token_count > budget_tokens: + truncated = True + break + entries.append( + ContextShowEntry( + path=frag.resource_id or frag.fragment_id, + size=frag.token_count, + tier=frag.tier.value, + fragment_id=frag.fragment_id, + ) + ) + running_tokens += frag.token_count + + return ContextShowResult( + entries=entries, + total_tokens=running_tokens, + budget_tokens=budget_tokens, + truncated=truncated, + ) -@app.command("show") -def acms_context_show( +def _apply_clear_filters( + tier_service: ContextTierService, + project_names: list[str], + path_filter: str | None, + tag_filter: str | None, + tier_filter: str | None, +) -> ContextClearResult: + """Remove fragments matching the given filters from the tier service. + + At least one filter must be provided; if none are provided, no + fragments are removed and the result reflects the current state. + + Args: + tier_service: The ACMS context tier service. + project_names: Project names to scope the query. + path_filter: Remove fragments whose resource_id matches this path. + tag_filter: Remove fragments whose metadata contains this tag key. + tier_filter: Remove fragments in this tier (hot/warm/cold). + + Returns: + A :class:`ContextClearResult` with counts and applied filters. + """ + filters_applied: dict[str, str | None] = { + "path": path_filter, + "tag": tag_filter, + "tier": tier_filter, + } + + # No filter -> nothing to remove + if path_filter is None and tag_filter is None and tier_filter is None: + if project_names: + remaining = len(tier_service.get_scoped_view(project_names)) + else: + remaining = len(tier_service.get_all_fragments()) + return ContextClearResult( + removed_count=0, + remaining_count=remaining, + filters_applied=filters_applied, + ) + + # Collect candidates + if project_names: + candidates = tier_service.get_scoped_view(project_names) + else: + candidates = tier_service.get_all_fragments() + + to_remove: list[TieredFragment] = [] + for frag in candidates: + match = False + if path_filter is not None and path_filter in (frag.resource_id or ""): + match = True + if tag_filter is not None and tag_filter in frag.metadata: + match = True + if tier_filter is not None and frag.tier == ContextTier(tier_filter): + match = True + if match: + to_remove.append(frag) + + # Remove matched fragments + removed_count = tier_service.remove_by_filter([f.fragment_id for f in to_remove]) + + # Count remaining + if project_names: + remaining = len(tier_service.get_scoped_view(project_names)) + else: + remaining = len(tier_service.get_all_fragments()) + + return ContextClearResult( + removed_count=removed_count, + remaining_count=remaining, + filters_applied=filters_applied, + ) + + +# --------------------------------------------------------------------------- +# CLI commands +# --------------------------------------------------------------------------- + + +@app.command(name="show") +def acms_show( view: Annotated[ str, - typer.Argument(help="View name/project to show assembled context for"), - ], + typer.Argument( + help=( + "Named view to assemble context for: " + "default, strategize, execute, or apply" + ) + ), + ] = "default", + project: Annotated[ + list[str] | None, + typer.Option( + "--project", + "-p", + help="Project name(s) to scope the query (repeatable)", + ), + ] = None, + budget: Annotated[ + int, + typer.Option( + "--budget", + help="Token budget for utilization computation", + ), + ] = _DEFAULT_BUDGET_TOKENS, + output_format: Annotated[ + str, + typer.Option("--format", "-f", help=_FORMAT_HELP), + ] = "rich", ) -> None: - """Display the assembled context for a specific ACMS view. + """Show assembled context for a named view with budget utilization. - Shows the context fragments stored in the hot/warm/cold tiers for - the given project view, including token budget utilization summary. + Displays the ACMS index entries for the given view, including: + + - Entry list (path, size in tokens, tier) + - Total size in tokens + - Budget utilization percentage + - Truncated flag (set when entries exceed the budget) + + Examples:: + + # Show default view context + agents acms show + + # Show strategize view for a specific project + agents acms show strategize --project local/myproject + + # Show in JSON format + agents acms show --format json + + # Show with custom budget + agents acms show --budget 4096 """ - console = _get_console() - - if not view or not view.strip(): - console.print("[red]Error:[/red] View name must not be empty.") - raise typer.Exit(code=1) - - try: - container = get_container() - tier_service = container.context_tier_service() - - # Get fragments scoped to the view (project name) - fragments = tier_service.get_scoped_view([view.strip()]) - - if not fragments: - console.print(f"[yellow]No context found for view: {view}[/yellow]") - return - - # Display assembled context - console.print(f"\n[bold]Assembled Context for View: {view}[/bold]") - - table = Table(title=f"Context Entries ({len(fragments)} total)") - table.add_column("Resource", style="cyan") - table.add_column("Tier", style="green") - table.add_column("Tokens", style="magenta") - table.add_column("Project", style="yellow") - - total_tokens = 0 - hot_tokens = 0 - warm_tokens = 0 - cold_tokens = 0 - for fragment in fragments: - resource_id, tier_label, token_str, project = _format_fragment_row(fragment) - total_tokens += fragment.token_count - if fragment.tier == ContextTier.HOT: - hot_tokens += fragment.token_count - elif fragment.tier == ContextTier.WARM: - warm_tokens += fragment.token_count - elif fragment.tier == ContextTier.COLD: - cold_tokens += fragment.token_count - table.add_row(resource_id, tier_label, token_str, project) - - console.print(table) - console.print(f"\n[bold]Total Tokens:[/bold] {total_tokens:,}") - - # Display tier metrics using actual token counts per tier - metrics = tier_service.get_metrics() - budget = tier_service.budget - hot_utilization = _format_budget_utilization( - hot_tokens, budget.max_tokens_hot or 1 + if view not in _VALID_VIEWS: + err_console.print( + f"[red]Invalid view '{view}': must be one of {sorted(_VALID_VIEWS)}[/red]" ) - warm_utilization = _format_budget_utilization( - metrics.warm_count, budget.max_decisions_warm or 1 - ) - cold_utilization = _format_budget_utilization( - metrics.cold_count, budget.max_decisions_cold or 1 - ) - console.print("\n[bold]Budget Utilization:[/bold]") + raise typer.Exit(1) + + from cleveragents.application.container import get_container + + container = get_container() + tier_service: ContextTierService = container.context_tier_service() + + project_names: list[str] = list(project) if project else [] + result = _assemble_show_result(tier_service, project_names, budget) + + if output_format.lower() == "json": + console.print(json.dumps(result.to_dict(), indent=2)) + return + + # Rich output + title = f"ACMS Context: {view} view" + if project_names: + title += f" ({', '.join(project_names)})" + + if not result.entries: console.print( - f" Hot tier: {metrics.hot_count} fragments" - f" ({hot_tokens:,} tokens)" - f" — {hot_utilization} of token budget" - ) - console.print( - f" Warm tier: {metrics.warm_count} decisions" - f" ({warm_tokens:,} tokens)" - f" — {warm_utilization} of decision budget" - ) - console.print( - f" Cold tier: {metrics.cold_count} decisions" - f" ({cold_tokens:,} tokens)" - f" — {cold_utilization} of decision budget" + Panel( + "[yellow]No context entries found.[/yellow]", + title=title, + expand=False, + ) ) + return - except CleverAgentsError as e: - _logger.exception("ACMS context show failed for view %r", view) - console = _get_console() - console.print( - "[red]Error:[/red] Failed to retrieve context." - " Please check logs for details." - ) - raise typer.Exit(code=1) from e - except Exception as e: - _logger.exception("Unexpected error in acms_context_show for view %r", view) - console = _get_console() - console.print( - f"[red]Error:[/red] Failed to retrieve context for view '{view}'." - ) - raise typer.Exit(code=1) from e + table = Table(title=title, expand=False) + table.add_column("#", style="dim", justify="right") + table.add_column("Path", style="cyan") + table.add_column("Size (tokens)", justify="right", style="magenta") + table.add_column("Tier", style="green") + + for idx, entry in enumerate(result.entries, 1): + table.add_row(str(idx), entry.path, str(entry.size), entry.tier) + + console.print(table) + + # Budget utilization summary line + truncated_str = " [yellow](truncated)[/yellow]" if result.truncated else "" + console.print( + f"Total: [bold]{result.total_tokens}[/bold] tokens | " + f"Budget: [bold]{result.budget_tokens}[/bold] tokens | " + f"Used: [bold]{result.budget_used_pct:.1f}%[/bold]{truncated_str}" + ) -@app.command("clear") -def acms_context_clear( - path: Annotated[ +@app.command(name="clear") +def acms_clear( + path_filter: Annotated[ str | None, - typer.Option("--path", help="Filter by resource path pattern (glob)"), + typer.Option( + "--path", + help="Remove entries whose path matches this value", + ), ] = None, - tag: Annotated[ + tag_filter: Annotated[ str | None, - typer.Option("--tag", help="Filter by metadata tag value"), + typer.Option( + "--tag", + help="Remove entries that have this metadata tag key", + ), ] = None, - tier: Annotated[ + tier_filter: Annotated[ str | None, typer.Option( "--tier", - help="Filter by tier (hot, warm, cold)", + help="Remove entries in this tier: hot, warm, or cold", + ), + ] = None, + project: Annotated[ + list[str] | None, + typer.Option( + "--project", + "-p", + help="Project name(s) to scope the query (repeatable)", ), ] = None, yes: Annotated[ - bool, typer.Option("--yes", "-y", help="Skip confirmation prompt") + bool, + typer.Option( + "--yes", + "-y", + help="Skip confirmation prompt", + ), ] = False, + output_format: Annotated[ + str, + typer.Option("--format", "-f", help=_FORMAT_HELP), + ] = "rich", ) -> None: - """Remove stale context entries from the ACMS index. + """Remove ACMS index entries by path, tag, or tier filter. - Removes context fragments by path pattern, metadata tag, or tier. - Requires confirmation unless --yes flag is provided. + Removes stale or unwanted entries from the ACMS context index. + At least one filter (``--path``, ``--tag``, or ``--tier``) must be + provided; without a filter, no entries are removed. + + A confirmation prompt is shown before deletion unless ``--yes`` is + passed. + + Examples:: + + # Remove entries matching a specific path + agents acms clear --path src/old_module.py + + # Remove all hot-tier entries + agents acms clear --tier hot + + # Remove entries tagged as stale + agents acms clear --tag stale + + # Skip confirmation + agents acms clear --path src/old.py --yes + + # Output result as JSON + agents acms clear --path src/old.py --yes --format json """ - console = _get_console() - - try: - container = get_container() - tier_service = container.context_tier_service() - - # Get all fragments and apply filters - all_fragments = tier_service.get_all_fragments() - - # Warn if no filters specified (clear ALL) - if not path and not tag and not tier: - console.print( - "[yellow]Warning:[/yellow] No filters specified. " - "This will remove ALL context entries." - ) - - # Apply filters - entries_to_remove = _filter_fragments(all_fragments, path, tag, tier) - - if not entries_to_remove: - console.print( - "[yellow]No context entries match the specified filters.[/yellow]" - ) - return - - # Warn on overly broad glob patterns that match many entries - if path and len(entries_to_remove) > _BROAD_PATTERN_THRESHOLD: - console.print( - f"[yellow]Warning:[/yellow] The path pattern '{path}' matches " - f"{len(entries_to_remove)} entries." - " Consider using a more specific pattern." - ) - - # Show what will be removed - console.print( - f"\n[bold]Context entries to remove " - f"({len(entries_to_remove)} total):[/bold]" - ) - - table = Table() - table.add_column("Resource", style="cyan") - table.add_column("Tier", style="green") - table.add_column("Project", style="yellow") - - for fragment in entries_to_remove[:10]: # Show first 10 - resource_id, tier_label, _, project = _format_fragment_row(fragment) - table.add_row(resource_id, tier_label, project) - - console.print(table) - - if len(entries_to_remove) > 10: - console.print(f" ... and {len(entries_to_remove) - 10} more entries") - - # Confirm if needed — handle cancellation before the main try/except - # to avoid catching typer.Exit from the confirmation prompt - if not yes: - confirmed = typer.confirm( - f"\nRemove {len(entries_to_remove)} context entries?" - ) - if not confirmed: - console.print("[yellow]Cancelled.[/yellow]") - return - - # Remove entries by evicting from their respective tiers - removed_count = _remove_fragments(tier_service, entries_to_remove) - console.print(f"\n[green]✓[/green] Removed {removed_count} context entries.") - - except CleverAgentsError as e: - _logger.exception("ACMS context clear failed") - console = _get_console() - console.print( - "[red]Error:[/red] Failed to clear context entries." - " Please check logs for details." - ) - raise typer.Exit(code=1) from e - except Exception as e: - _logger.exception("Unexpected error in acms_context_clear") - console = _get_console() - console.print("[red]Error:[/red] Failed to clear context entries.") - raise typer.Exit(code=1) from e - - -def _filter_fragments( - fragments: list[TieredFragment], - path: str | None, - tag: str | None, - tier_filter: str | None, -) -> list[TieredFragment]: - """Filter fragments by path pattern, tag, and/or tier. - - Args: - fragments: All fragments to filter. - path: Glob pattern to match against resource_id. - tag: Tag value to match in fragment metadata. - tier_filter: Tier name to filter by (hot, warm, cold). - - Returns: - Filtered list of fragments matching all specified criteria. - """ - result = fragments - - if path: - result = [f for f in result if fnmatch.fnmatch(f.resource_id, path)] - - if tag: - result = [ - f - for f in result - if f.metadata.get("tag") == tag or f.metadata.get("tags") == tag - ] - - if tier_filter: + if tier_filter is not None: try: - target_tier = ContextTier(tier_filter.lower()) - result = [f for f in result if f.tier == target_tier] + ContextTier(tier_filter) except ValueError: - # Invalid tier name — return empty (no matches) - return [] + err_console.print( + f"[red]Invalid tier '{tier_filter}': must be hot, warm, or cold[/red]" + ) + raise typer.Exit(1) from None - return result + from cleveragents.application.container import get_container + container = get_container() + tier_service: ContextTierService = container.context_tier_service() -def _remove_fragments( - tier_service: _TierServiceProtocol, - fragments: list[TieredFragment], -) -> int: - """Remove the given fragments from the tier service. + project_names: list[str] = list(project) if project else [] - Removes fragments by calling _remove_from_all on the tier service. - ContextTierService is thread-safe (RLock-protected); manual locking - here would duplicate internal synchronization and risk mismatches. + # No filter -> nothing to do + if path_filter is None and tag_filter is None and tier_filter is None: + if output_format.lower() == "json": + result = ContextClearResult( + removed_count=0, + remaining_count=len( + tier_service.get_scoped_view(project_names) + if project_names + else tier_service.get_all_fragments() + ), + filters_applied={"path": None, "tag": None, "tier": None}, + ) + console.print(json.dumps(result.to_dict(), indent=2)) + else: + console.print( + "[yellow]No filter specified. " + "Use --path, --tag, or --tier to select entries.[/yellow]" + ) + return - Args: - tier_service: The ContextTierService instance (or compatible implementation). - fragments: Fragments to remove. + # Confirmation prompt + if not yes: + filter_parts: list[str] = [] + if path_filter: + filter_parts.append(f"path={path_filter!r}") + if tag_filter: + filter_parts.append(f"tag={tag_filter!r}") + if tier_filter: + filter_parts.append(f"tier={tier_filter!r}") + filter_desc = ", ".join(filter_parts) + confirmed = typer.confirm(f"Remove ACMS entries matching {filter_desc}?") + if not confirmed: + console.print("[yellow]Cancelled.[/yellow]") + raise typer.Exit(0) - Returns: - Number of fragments successfully removed. - """ - removed = 0 - for fragment in fragments: - tier_service._remove_from_all(fragment.fragment_id) - removed += 1 + result = _apply_clear_filters( + tier_service, + project_names, + path_filter, + tag_filter, + tier_filter, + ) - return removed + if output_format.lower() == "json": + console.print(json.dumps(result.to_dict(), indent=2)) + return - -@app.callback(invoke_without_command=True) -def acms_context_default(ctx: typer.Context) -> None: - """Show ACMS context information when no subcommand is provided.""" - if ctx.invoked_subcommand is None: - console = _get_console() - console.print("[bold]ACMS Context Management[/bold]") - console.print("\nAvailable commands:") - console.print(" show - Display assembled context for a view") - console.print(" clear - Remove context entries with filtering") - console.print( - "\nUse 'agents acms context --help' for more information." - ) + # Rich output + console.print( + f"[green]✓[/green] Removed [bold]{result.removed_count}[/bold] " + f"entries. Remaining: [bold]{result.remaining_count}[/bold]." + ) diff --git a/src/cleveragents/cli/main.py b/src/cleveragents/cli/main.py index dfb584e8a..724d0e22c 100644 --- a/src/cleveragents/cli/main.py +++ b/src/cleveragents/cli/main.py @@ -237,6 +237,11 @@ def _register_subcommands() -> None: name="repo", help="Repository indexing management", ) + app.add_typer( + acms_context.app, + name="acms", + help="ACMS context index commands (show assembled context, clear entries)", + ) _subcommands_registered = True