feat(acms): implement context show and context clear CLI commands #10780
@@ -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"
|
||||
@@ -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
|
||||
@@ -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}"
|
||||
@@ -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"
|
||||
@@ -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
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@@ -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 <view>`` and ``agents acms clear`` to
|
||||
inspect assembled context entries and remove stale or unwanted entries
|
||||
from the ACMS index.
|
||||
|
||||
Canonical path: ``agents acms context <subcommand>``
|
||||
Commands:
|
||||
- ``agents acms show <view>`` — 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 <command>`` (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 <view> - Display assembled context for a view")
|
||||
console.print(" clear - Remove context entries with filtering")
|
||||
console.print(
|
||||
"\nUse 'agents acms context <command> --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]."
|
||||
)
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user